kfsclient.cc

nandflash文件系统源代码

    mCwd = path;
    return 0;
}

///
/// To allow tools to get at "pwd"
///
string
KfsClientImpl::GetCwd()
{
    return mCwd;
}


///
/// Make a directory hierarchy in KFS.
///
int
KfsClientImpl::Mkdirs(const char *pathname)
{
    MutexLock l(&mMutex);

    int res;
    string path = pathname;
    string component;
    const char slash = '/';
    string::size_type startPos = 1, endPos;
    bool done = false;

    //
    // Walk from the root down to the last part of the path making the
    // directory hierarchy along the way.  If any of the components of
    // the path is a file, error out.
    //
    while (!done) {
        endPos = path.find(slash, startPos);
        if (endPos == string::npos) {
            done = true;
            component = pathname;
        } else {
            component = path.substr(0, endPos);
            startPos = endPos + 1;
        }
	if (Exists(component.c_str())) {
	    if (IsFile(component.c_str()))
		return -ENOTDIR;
	    continue;
	}
	res = Mkdir(component.c_str());
        if (res == -EEXIST) {
            // when there are multiple clients trying to make the same
            // directory hierarchy, the first one wins; the subsequent
            // ones get a EEXIST and that is not fatal
            continue;
        }
	if (res < 0)
	    return res;
    }

    return 0;
}
 
///
/// Make a directory in KFS.
/// @param[in] pathname		The full pathname such as /.../dir
/// @retval 0 if mkdir is successful; -errno otherwise
int
KfsClientImpl::Mkdir(const char *pathname)
{
    MutexLock l(&mMutex);

    kfsFileId_t parentFid;
    string dirname;
    int res = GetPathComponents(pathname, &parentFid, dirname);
    if (res < 0)
	return res;

    MkdirOp op(nextSeq(), parentFid, dirname.c_str());
    DoMetaOpWithRetry(&op);
    if (op.status < 0) {
	return op.status;
    }

    // Everything is good now...
    int fte = ClaimFileTableEntry(parentFid, dirname.c_str(), pathname);
    if (fte < 0)	// Too many open files
	return -EMFILE;

    mFileTable[fte]->fattr.fileId = op.fileId;
    // setup the times and such
    mFileTable[fte]->fattr.Init(true);

    return 0;
}

///
/// Remove a directory in KFS.
/// @param[in] pathname		The full pathname such as /.../dir
/// @retval 0 if rmdir is successful; -errno otherwise
int
KfsClientImpl::Rmdir(const char *pathname)
{
    MutexLock l(&mMutex);

    string dirname;
    kfsFileId_t parentFid;
    int res = GetPathComponents(pathname, &parentFid, dirname);
    if (res < 0)
	return res;

    int fte = LookupFileTableEntry(parentFid, dirname.c_str());
    if (fte > 0)
	ReleaseFileTableEntry(fte);

    RmdirOp op(nextSeq(), parentFid, dirname.c_str());
    (void)DoMetaOpWithRetry(&op);
    return op.status;
}

///
/// Remove a directory hierarchy in KFS.
/// @param[in] pathname		The full pathname such as /.../dir
/// @retval 0 if rmdir is successful; -errno otherwise
int
KfsClientImpl::Rmdirs(const char *pathname)
{
    MutexLock l(&mMutex);

    vector<KfsFileAttr> entries;
    int res;

    if ((res = ReaddirPlus(pathname, entries, false)) < 0)
        return res;

    string dirname = pathname;

    int len = dirname.size();
    if (dirname[len - 1] == '/')
        dirname.erase(len - 1);

    for (size_t i = 0; i < entries.size(); i++) {
        if ((entries[i].filename == ".") || (entries[i].filename == ".."))
            continue;

        string d = dirname;
        
        d += "/" + entries[i].filename;
        if (entries[i].isDirectory) {
            res = Rmdirs(d.c_str());
        } else {
            res = Remove(d.c_str());
        }
        if (res < 0)
            break;
    }
    if (res < 0)
        return res;

    res = Rmdir(pathname);

    return res;

}

///
/// Read a directory's contents.  This is analogous to READDIR in
/// NFS---just reads the directory contents and returns the names;
/// you'll need to lookup the attributes next.  The resulting
/// directory entries are sorted lexicographically.
///
/// XXX NFS READDIR also returns the file ids, and we should do
/// the same here.
///
/// @param[in] pathname	The full pathname such as /.../dir
/// @param[out] result	The filenames in the directory
/// @retval 0 if readdir is successful; -errno otherwise
int
KfsClientImpl::Readdir(const char *pathname, vector<string> &result)
{
    MutexLock l(&mMutex);

    int fte = LookupFileTableEntry(pathname);
    if (fte < 0) {
	// open the directory for reading
	fte = Open(pathname, O_RDONLY);
    }

    if (fte < 0)
	return fte;

    if (!mFileTable[fte]->fattr.isDirectory)
	return -ENOTDIR;

    kfsFileId_t dirFid = mFileTable[fte]->fattr.fileId;

    ReaddirOp op(nextSeq(), dirFid);
    DoMetaOpWithRetry(&op);
    int res = op.status;
    if (res < 0)
	return res;

    istringstream ist;
    char filename[MAX_FILENAME_LEN];
    assert(op.contentBuf != NULL);
    ist.str(op.contentBuf);
    result.resize(op.numEntries);
    for (int i = 0; i < op.numEntries; ++i) {
	// ist >> result[i];
	ist.getline(filename, MAX_FILENAME_LEN);
	result[i] = filename;
        // KFS_LOG_VA_DEBUG("Entry: %s", filename);
    }
    sort(result.begin(), result.end());
    return res;
}

///
/// Read a directory's contents and get the attributes.  This is
/// analogous to READDIRPLUS in NFS.  The resulting directory entries
/// are sort lexicographically.
///
/// @param[in] pathname	The full pathname such as /.../dir
/// @param[out] result	The filenames in the directory and their attributes
/// @retval 0 if readdir is successful; -errno otherwise
int
KfsClientImpl::ReaddirPlus(const char *pathname, vector<KfsFileAttr> &result,
                           bool computeFilesize)
{
    MutexLock l(&mMutex);

    int fte = LookupFileTableEntry(pathname);
    if (fte < 0)	 // open the directory for reading
	fte = Open(pathname, O_RDONLY);
    if (fte < 0)
	   return fte;

    FileAttr *fa = FdAttr(fte);
    if (!fa->isDirectory)
	return -ENOTDIR;

    kfsFileId_t dirFid = fa->fileId;

    ReaddirPlusOp op(nextSeq(), dirFid);
    (void)DoMetaOpWithRetry(&op);
    int res = op.status;
    if (res < 0) {
	return res;
    }

    vector<FileChunkInfo> fileChunkInfo;
    istringstream ist;
    string entryInfo;
    boost::scoped_array<char> line;
    int count = 0, linelen = 1 << 20, numchars;
    const string entryDelim = "Begin-entry";
    string s(op.contentBuf, op.contentLength);

    ist.str(s);

    KFS_LOG_VA_DEBUG("# of entries: %d", op.numEntries);

    line.reset(new char[linelen]);

    // the format is:
    // Begin-entry <values> Begin-entry <values>
    // the last entry doesn't have a end-marker
    while (count < op.numEntries) {
        ist.getline(line.get(), linelen);

        numchars = ist.gcount();
        if (numchars != 0) {
            if (line[numchars - 2] == '\r')
                line[numchars - 2] = '\0';

            KFS_LOG_VA_DEBUG("entry: %s", line.get());

            if (line.get() != entryDelim) {
                entryInfo += line.get();
                entryInfo += "\r\n";
                continue;
            }
            // we hit a delimiter; if this is the first one, we
            // continue so that we can build up the key/value pairs
            // for the entry.
            if (entryInfo == "")
                continue;
        }
        count++;
        // sanity
        if (entryInfo == "")
            continue;

        // previous entry is all done...process it
        Properties prop;
        KfsFileAttr fattr;
        string s;
        istringstream parserStream(entryInfo);
        const char separator = ':';

        prop.loadProperties(parserStream, separator, false);
        fattr.filename = prop.getValue("Name", "");
        fattr.fileId = prop.getValue("File-handle", -1);
        s = prop.getValue("Type", "");
        fattr.isDirectory = (s == "dir");

        s = prop.getValue("M-Time", "");
        GetTimeval(s, fattr.mtime);

        s = prop.getValue("C-Time", "");
        GetTimeval(s, fattr.ctime);

        s = prop.getValue("CR-Time", "");
        GetTimeval(s, fattr.crtime);

        entryInfo = "";

        fattr.numReplicas = prop.getValue("Replication", 1);
        fattr.fileSize = prop.getValue("File-size", (off_t) -1);
        if (fattr.fileSize != -1) {
            KFS_LOG_VA_DEBUG("Got file size from server for %s: %lld", 
                             fattr.filename.c_str(), fattr.fileSize);
        }

        // get the location info for the last chunk
        FileChunkInfo lastChunkInfo(fattr.filename, fattr.fileId);

        lastChunkInfo.lastChunkOffset = prop.getValue("Chunk-offset", 0);
        lastChunkInfo.chunkCount = prop.getValue("Chunk-count", 0);
        lastChunkInfo.numReplicas = prop.getValue("Replication", 1);
        lastChunkInfo.cattr.chunkId = prop.getValue("Chunk-handle", (kfsFileId_t) -1);
        lastChunkInfo.cattr.chunkVersion = prop.getValue("Chunk-version", (int64_t) -1);

        int numReplicas = prop.getValue("Num-replicas", 0);
        string replicas = prop.getValue("Replicas", "");

        if (replicas != "") {
            istringstream ser(replicas);
            ServerLocation loc;

            for (int i = 0; i < numReplicas; ++i) {
                ser >> loc.hostname;
                ser >> loc.port;
                lastChunkInfo.cattr.chunkServerLoc.push_back(loc);
            }
        }
        fileChunkInfo.push_back(lastChunkInfo);
        result.push_back(fattr);
    }

    if (computeFilesize) {

        for (uint32_t i = 0; i < result.size(); i++) {
            if ((fileChunkInfo[i].chunkCount == 0) || (result[i].isDirectory)) {
                result[i].fileSize = 0;
                continue;
            }

            int fte = LookupFileTableEntry(dirFid, result[i].filename.c_str());

            if (fte >= 0) {
                result[i].fileSize = mFileTable[fte]->fattr.fileSize;
            } 
        }
        ComputeFilesizes(result, fileChunkInfo);

        for (uint32_t i = 0; i < result.size(); i++) 
            if (result[i].fileSize < 0)
                result[i].fileSize = 0;

    }

    // if there are too many entries in the dir, then the caller is
    // probably scanning the directory.  don't put it in the cache
    if (result.size() > 128) {
        sort(result.begin(), result.end());
        return res;
    }

    string dirname = build_path(mCwd, pathname);
    string::size_type len = dirname.size();
    if ((len > 0) && (dirname[len - 1] == '/'))
        dirname.erase(len - 1);
    
    for (uint32_t i = 0; i < result.size(); i++) {
        int fte = LookupFileTableEntry(dirFid, result[i].filename.c_str());

        if (fte >= 0) {
            // if we computed the filesize, then we stash it; otherwise, we'll
            // set the value to -1 and force a recompute later...
            mFileTable[fte]->fattr.fileSize = result[i].fileSize;
            continue;
        }

        if (fte < 0) {
            string fullpath;
            if ((result[i].filename == ".") || (result[i].filename == ".."))
                fullpath = "";
            else
                fullpath = dirname + "/" + result[i].filename;

            fte = AllocFileTableEntry(dirFid, result[i].filename.c_str(), fullpath);
            if (fte < 0)
                continue;
        }

        mFileTable[fte]->fattr.fileId = result[i].fileId;
        mFileTable[fte]->fattr.mtime = result[i].mtime;
        mFileTable[fte]->fattr.ctime = result[i].ctime;
        mFileTable[fte]->fattr.ctime = result[i].crtime;
        mFileTable[fte]->fattr.isDirectory = result[i].isDirectory;
        mFileTable[fte]->fattr.chunkCount = fileChunkInfo[i].chunkCount;
        mFileTable[fte]->fattr.numReplicas = fileChunkInfo[i].numReplicas;

        mFileTable[fte]->openMode = 0;
        // if we computed the filesize, then we stash it; otherwise, we'll
        // set the value to -1 and force a recompute later...
        mFileTable[fte]->fattr.fileSize = result[i].fileSize;
    }

    sort(result.begin(), result.end());

    return res;
}

///
/// Do du on the metaserver side; much faster than recursive traversal.
///
/// @param[in] pathname	The full pathname such as /.../dir
/// @param[out] numFiles # of files in the directory tree
/// @param[out] numBytes # of bytes used by the directory tree
/// @retval 0 if getdirsummary is successful; -errno otherwise
int
KfsClientImpl::GetDirSummary(const char *pathname, uint64_t &numFiles, uint64_t &numBytes)
{
    MutexLock l(&mMutex);

    int fte = LookupFileTableEntry(pathname);
    if (fte < 0)	 // open the directory for reading
	fte = Open(pathname, O_RDONLY);
    if (fte < 0)
	   return fte;

    FileAttr *fa = FdAttr(fte);
    if (!fa->isDirectory)
	return -ENOTDIR;

    kfsFileId_t dirFid = fa->fileId;

    GetDirSummaryOp op(nextSeq(), dirFid);
    (void)DoMetaOpWithRetry(&op);
    int res = op.status;
    if (res < 0) {
	return res;
    }
    numFiles = op.numFiles;
    numBytes = op.numBytes;
    return 0;
}

int
KfsClientImpl::Stat(const char *pathname, struct stat &result, bool computeFilesize)
{
    MutexLock l(&mMutex);

    KfsFileAttr kfsattr;

    int fte = LookupFileTableEntry(pathname);

    if (fte >= 0) {
	kfsattr = mFileTable[fte]->fattr;
    }

    // either we don't have the attributes cached or it is a file and
    // we are asked to compute the size and we don't know the size,
    // lookup the attributes
    if ((fte < 0) || ((!kfsattr.isDirectory) && computeFilesize && 
                      (kfsattr.fileSize < 0))) {
	kfsFileId_t parentFid;
	string filename;
	int res = GetPathComponents(pathname, &parentFid, filename);
	if (res == 0) {
	    res = LookupAttr(parentFid, filename.c_str(), kfsattr, computeFilesize);
        }
	if (res < 0)
	    return res;
    }


    memset(&result, 0, sizeof (struct stat));
    result.st_mode = kfsattr.isDirectory ? S_IFDIR : S_IFREG;
    result.st_size = kfsattr.fileSize;
    result.st_atime = kfsattr.crtime.tv_sec;
    result.st_mtime = kfsattr.mtime.tv_sec;
    result.st_ctime = kfsattr.ctime.tv_sec;
    return 0;
}

bool
KfsClientImpl::Exists(const char *pathname)
{
    MutexLock l(&mMutex);

    struct stat dummy;

    return Stat(pathname, dummy, false) == 0;