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

---

FileSystem.h

Go to the documentation of this file.

#ifndef Impala_Persistency_FileSystem_h
#define Impala_Persistency_FileSystem_h

#ifdef unix
#include <sys/stat.h>
#include <sys/types.h>
#else
#include <direct.h>
#include <errno.h>
#endif

#include <vector>

#include "Basis/FileName.h"
#include "Basis/FileNameTmp.h"
#include "Util/ChannelPool.h"
#include "Util/ReadIOBufferFromChannel.h"
#include "Util/IOBufferFile.h"
#include "Util/IOBufferChannel.h"

namespace Impala
{
namespace Persistency
{


class FileSystem
{
public:

    FileSystem(CString data, CString dataServer)
    {
        CmdOptions& options = CmdOptions::GetInstance();
        mDataString = data;
        mData = StringList(mDataString, ';');
        mDataServer = dataServer;
        mDataChannel = 0;
        if (!mDataServer.empty())
        {
            ILOG_DEBUG_HEADNODE("Retrieving channel to server " << mDataServer);
            String passFile = options.GetString("passwordFile");
            mDataChannel = Util::ChannelPool::Instance().Get(mDataServer,
                                                             passFile);
        }
        mOverride = options.GetBool("override");
#ifdef WIN32
        // printf("%d\n", _getmaxstdio()); // normally 512
        _setmaxstdio(2048); // set it to the absolute maximum
#endif
    }

    bool
    Valid() const
    {
        return (mDataServer.empty() || mDataChannel);
    }

    // For use in data server only : reset data path for new request.
    void
    SetDataPath(String data)
    {
        mDataString = data;
        mData.clear();
        mData = StringList(mDataString, ';');
    }

    String
    GetDataPath()
    {
        return mDataString;
    }

    // For use in data server only : reset override for new request.
    void
    SetOverride(bool override)
    {
        mOverride = override;
    }

    // Directory part

    String
    MakeDir(CString dir)
    {
        return DoMkDirPath(dir);
    }

    void
    DeleteDir(CString dir)
    {
        if (mDataChannel)
        {
            char* buf = mDataChannel->Buffer();
            sprintf(buf, "deletedir \"%s\" \"%s\"\0", dir.c_str(),
                    mDataString.c_str());
            mDataChannel->SendRequest(strlen(buf)+1);
        }
        else
        {
            DoDeleteDir(dir);
        }
    }

    // File part

    String
    GetFilePath(CString file, bool toWrite, bool silent)
    {
        if (! Valid())
            return "";

        CString path = file;
        if (toWrite)
            return GetWritableFile(mData.begin(), mData.end(), path, silent);
        return GetReadableFile(mData.begin(), mData.end(), path, silent);
    }

    String
    GetFilePath(CString dir, CString file, bool toWrite, bool silent)
    {
        if (! Valid())
            return "";

        String path = FileNameConcat(dir, file);
        if (toWrite)
            return GetWritableFile(mData.begin(), mData.end(), path, silent);
        return GetReadableFile(mData.begin(), mData.end(), path, silent);
    }

    void
    DeleteFile(CString path)
    {
        if (mDataChannel)
        {
            char* buf = mDataChannel->Buffer();
            sprintf(buf, "deletefile \"%s\"\0", path.c_str());
            mDataChannel->SendRequest(strlen(buf)+1);
        }
        else
        {
            unlink(path.c_str());
        }
    }

    Util::IOBuffer*
    GetIOBuffer(CString fileName, bool readMode, bool useMemoryIfLocal,
                String useLocalFileIfRemote, Int64 useLocalSizeIfRemote = 0,
                bool useIOBufferChannel = false)
    {
        Util::IOBuffer* res = 0;
        ILOG_DEBUG("GetIOBuffer [" << fileName << "]");
        if (! Valid())
            return res;

        if (mDataChannel)
        {
            if (readMode)
            {
                if (useIOBufferChannel)
                {
                    ILOG_DEBUG("GetIOBuffer: Read using Channel to " <<
                               mDataServer << ": " << fileName);
                    res = new Util::IOBufferChannel(fileName, true,
                                                    mDataChannel);
                }
                else
                {
                    ILOG_DEBUG("GetIOBuffer: Read using File: " << fileName);
                    /* the race condition that occurs when someone passes in
                       "tmp" while reading a file is very difficult to track
                       down (only happens with MPI-jobs). Therefore, generate 
                       a temporary filename even in read mode */
                    if (useLocalFileIfRemote == "tmp")
                        useLocalFileIfRemote = FileNameTmp();
                    res = Util::ReadIOBufferFromChannel(mDataChannel, fileName,
                                                        useLocalFileIfRemote);
                }
            }
            else
            {
                if (useIOBufferChannel)
                {
                    ILOG_DEBUG("GetIOBuffer: Write using Channel " << fileName);
                    res = new Util::IOBufferChannel(fileName, false,
                                                    mDataChannel);
                }
                else if (useLocalFileIfRemote.empty())
                {
                    ILOG_DEBUG("GetIOBuffer: Write using mem " << fileName);
                    res = new Util::IOBuffer(useLocalSizeIfRemote);
                }
                else
                {
                    ILOG_DEBUG("GetIOBuffer: Write using File " << fileName);
                    if (useLocalFileIfRemote == "tmp")
                        useLocalFileIfRemote = FileNameTmp();
                    res = new Util::IOBufferFile(useLocalFileIfRemote, false,
                                                 false);
                }
                if (!useIOBufferChannel)
                    res->SetUseChannel(mDataChannel, fileName,
                                       useLocalFileIfRemote);
            }
        }
        else
        {
            res = new Util::IOBufferFile(fileName, readMode, useMemoryIfLocal);
        }

        if (res)
        {
            if (!res->Valid())
            {
                delete res;
                res = 0;
            }
        }
        return res;
    }

    Util::Channel*
    GetDataChannel()
    {
        return mDataChannel;
    }

    void
    Dump() const
    {
        ILOG_INFO_HEADNODE("Dump: data=[" << mDataString << "] server=[" <<
                       mDataServer << "]");
    }

private:

    // Utilities

    void
    DoMkDir(CString dir)
    {
#ifdef unix
        mode_t m = 0755;
        mkdir(dir.c_str(), m);
#else
        if ((_mkdir(StringReplaceAll(dir, "/", "\\").c_str()) != 0) && 
            (errno != EEXIST))
        {
            ILOG_ERROR("[DoMkDir] Failed to create directory: " << dir << 
                       " (" << strerror(errno) << ")");
        }
#endif
    }

    String
    DoMkDirPath(CString dir)
    {
        if (mDataChannel)
            return DoMkDirPathServer(dir);

        return DoMkDirPathLocal(ConcatDir(*mData.begin(), dir));
    }

    String
    DoMkDirPathLocal(CString dirWithBackSlashes)
    {
        String dir = StringReplaceAll(dirWithBackSlashes, "\\", "/");
        ILOG_DEBUG("DoMkDirPath: " << dir);

        String path = "";
        if (dir[0] == '/')
        {
#ifdef WIN32
            if (dir.find("//") == 0)
            {
                // construct path head upto and including fourth slash
                int headEndPos = 1;
                int nextSlashPos = dir.find("/", headEndPos + 1);
                if (nextSlashPos != String::npos)
                {
                    headEndPos = nextSlashPos;
                    nextSlashPos = dir.find("/", headEndPos + 1);
                    if (nextSlashPos != String::npos)
                        headEndPos = nextSlashPos;
                }
                path = dir.substr(0, headEndPos + 1);
                dir = dir.substr(headEndPos + 1);
            }
            else
#endif
                path = String("/");
        }
        
        StringList subDirs(dir, '/');
        for (StringListCI i=subDirs.begin() ; i!=subDirs.end() ; i++)
        {
            CString subDir = *i;
            ILOG_DEBUG("  DoMkDirPath: subDir = " << subDir);
            if (subDir.size() == 0 || subDir == ".")
                continue;
            if ((path.size() > 0) && (*path.rbegin() != '/'))
                path = path + "/";
            path = path + *i;
            DoMkDir(path);
        }
        return path;
    }

    String
    DoMkDirPathServer(CString dir)
    {
        char* buf = mDataChannel->Buffer();
        sprintf(buf, "domkdirpath \"%s\" \"%s\"\0", dir.c_str(),
                mDataString.c_str());
        int len = mDataChannel->SendRequest(strlen(buf)+1);
        if (mDataChannel->LastSendHadError())
            return "";
        buf[len] = 0;
        String res(buf);
        ILOG_DEBUG("Got dir [" << res << "]");
        return res;
    }

    void
    DoDeleteDir(String dir)
    {
        dir = ConcatDir(*mData.begin(), dir);
#ifdef WIN32
        dir = StringReplaceAll(dir, "/", "\\", true);
        String cmdLine("if exist " + dir + " rmdir /s/q " + dir);
#else
        while (StringEndsWith(dir, "/."))
            dir = dir.substr(0, dir.size()-2);
        String cmdLine("rm -rf " + dir);
#endif
        char sysBuf[2048];
        ILOG_INFO("[DoDeleteDir] " << cmdLine);
        sprintf(sysBuf, "%s", cmdLine.c_str());
        //
        int sysRes = system(sysBuf);
        if (sysRes != 0)
        {
            ILOG_ERROR("[DoDeleteDir] System call failed: ["  << sysBuf << "]");
        }
        //
    }

    String
    GetReadableFile(StringListCI begin, StringListCI end, CString file,
                    bool silent)
    {
        if (mDataChannel)
            return GetReadableFileServer(file, silent);
        return GetReadableFileLocal(begin, end, file, silent);
    }

    String
    GetReadableFileLocal(StringListCI begin, StringListCI end, CString file,
                         bool silent)
    {
        StringListCI i;
        for (i=begin ; i!=end ; i++)
        {
            String d = StringResolveEnv(*i);
            String path = FileNameConcat(d, file);
            ILOG_DEBUG("GetReadableFile: trying " << path);
            if (Util::IOBufferFile::FileExists(path))
                return path;
        }

        if (!silent)
        {
            ILOG_ERROR("Unable to find " << file << " in path ");
            for (i=begin ; i!=end ; i++)
                ILOG_INFO("    " << *i);
        }
        return String("");
    }

    String
    GetReadableFileServer(CString file, bool silent)
    {
        char* buf = mDataChannel->Buffer();
        sprintf(buf, "getreadablefile \"%s\" \"%s\" %d\0", file.c_str(),
                mDataString.c_str(), silent);
        int len = mDataChannel->SendRequest(strlen(buf)+1);
        if (mDataChannel->LastSendHadError())
            return "";
        buf[len] = 0;
        String res(buf);
        ILOG_DEBUG("Got file [" << res << "] from " << mDataServer);
        return res;
    }

    String
    GetWritableFile(StringListCI begin, StringListCI end, CString file,
                    bool silent)
    {
        if (mDataChannel)
            return GetWritableFileServer(file, silent);
        return GetWritableFileLocal(begin, end, file, silent);
    }

    String
    GetWritableFileLocal(StringListCI begin, StringListCI end, CString file,
                         bool silent)
    {
        if (begin == end)
        {
            ILOG_ERROR("GetWritableFile: no search path");
            return String("");
        }
        String d = StringResolveEnv(*begin);
        String path = FileNameConcat(d, file);
        ILOG_DEBUG("GetWritableFile: trying " << path);
        if (Util::IOBufferFile::FileExists(path))
        {
            if (! (mOverride || silent))   // silent implies "just checking"
            {
                if (!silent)
                    ILOG_ERROR(path << " already exists, will not override!");
                return String("");
            }
        }
        return path;
    }

    String
    GetWritableFileServer(CString file, bool silent)
    {
        char* buf = mDataChannel->Buffer();
        sprintf(buf, "getwritablefile \"%s\" \"%s\" %d %d\0", file.c_str(),
                mDataString.c_str(), silent, mOverride);
        int len = mDataChannel->SendRequest(strlen(buf)+1);
        if (mDataChannel->LastSendHadError())
            return "";
        buf[len] = 0;
        String res(buf);
        ILOG_DEBUG("Got file [" << res << "] from " << mDataServer);
        return res;
    }

    String
    ConcatDir(CString dir, CString subDir)
    {
        return (subDir.empty() ? dir : (dir.empty() ? subDir
                                                    : (dir + "/" + subDir)));
    }

    bool           mOverride;
    String         mDataString;
    StringList     mData;
    String         mDataServer;
    Util::Channel* mDataChannel;

    ILOG_VAR_DEC;
};

ILOG_VAR_INIT(FileSystem, Impala.Persistency);

} // namespace Persistency
} // namespace Impala

#endif

---