<commit_before>//---------------------------------------------------------- -*- Mode: C++ -*-
// $Id$ 
//
// Created 2006/04/18
// Author: <NAME>
//
// Copyright 2008 Quantcast Corp.
// Copyright 2006-2008 Kosmix Corp.
//
// This file is part of Kosmos File System (KFS).
//
// Licensed under the Apache License, Version 2.0
// (the "License"); you may not use this file except in compliance with
// the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
//
// \file KfsClient.cc
// \brief Kfs Client-library code.
//
//----------------------------------------------------------------------------

#include "KfsClient.h"
#include "KfsClientInt.h"

#include "common/config.h"
#include "common/properties.h"
#include "common/log.h"
#include "meta/kfstypes.h"
#include "libkfsIO/Checksum.h"
#include "Utils.h"

extern "C" {
#include <signal.h>
#include <stdlib.h>
}
#include <cerrno>
#include <iostream>
#include <string>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <boost/scoped_array.hpp>

using std::string;
using std::ostringstream;
using std::istringstream;
using std::min;
using std::max;
using std::map;
using std::vector;
using std::sort;
using std::transform;
using std::random_shuffle;

using std::cout;
using std::endl;

using namespace KFS;

const int CMD_BUF_SIZE = 1024;

// Set the default timeout for server I/O's to be 5 mins for now.
// This is intentionally large so that we can do stuff in gdb and not
// have the client timeout in the midst of a debug session.
struct timeval gDefaultTimeout = {300, 0};


namespace {
    Properties & theProps()
    {
        static Properties p;
        return p;
    }
}   

KfsClientFactory *
KFS::getKfsClientFactory()
{
    return KfsClientFactory::Instance();
}

KfsClientPtr
KfsClientFactory::GetClient(const char *propFile)
{
    bool verbose = false;
#ifdef DEBUG
    verbose = true;
#endif
    if (theProps().loadProperties(propFile, '=', verbose) != 0) {
        KfsClientPtr clnt;
	return clnt;
    }

    return GetClient(theProps().getValue("metaServer.name", ""),
                     theProps().getValue("metaServer.port", -1));

}

class MatchingServer {
    ServerLocation loc;
public:
    MatchingServer(const ServerLocation &l) : loc(l) { }
    bool operator()(KfsClientPtr &clnt) const {
        return clnt->GetMetaserverLocation() == loc;
    }
    bool operator()(const ServerLocation &other) const {
        return other == loc;
    }
};

KfsClientPtr
KfsClientFactory::GetClient(const std::string metaServerHost, int metaServerPort)
{
    vector<KfsClientPtr>::iterator iter;
    ServerLocation loc(metaServerHost, metaServerPort);

    iter = find_if(mClients.begin(), mClients.end(), MatchingServer(loc));
    if (iter != mClients.end())
        return *iter;

    KfsClientPtr clnt;

    clnt.reset(new KfsClient());

    clnt->Init(metaServerHost, metaServerPort);
    if (clnt->IsInitialized())
        mClients.push_back(clnt);
    else
        clnt.reset();

    return clnt;
}


KfsClient::KfsClient()
{
    mImpl = new KfsClientImpl();
}

KfsClient::~KfsClient()
{
    delete mImpl;
}

void
KfsClient::SetLogLevel(string logLevel)
{
    if (logLevel == "DEBUG")
        MsgLogger::SetLevel(log4cpp::Priority::DEBUG);
    else if (logLevel == "INFO")
        MsgLogger::SetLevel(log4cpp::Priority::INFO);
}

int 
KfsClient::Init(const std::string metaServerHost, int metaServerPort)
{
    return mImpl->Init(metaServerHost, metaServerPort);
}

bool 
KfsClient::IsInitialized()
{
    return mImpl->IsInitialized();
}

int
KfsClient::Cd(const char *pathname)
{
    return mImpl->Cd(pathname);
}

string
KfsClient::GetCwd()
{
    return mImpl->GetCwd();
}

int
KfsClient::Mkdirs(const char *pathname)
{
    return mImpl->Mkdirs(pathname);
}

int 
KfsClient::Mkdir(const char *pathname)
{
    return mImpl->Mkdir(pathname);
}

int 
KfsClient::Rmdir(const char *pathname)
{
    return mImpl->Rmdir(pathname);
}

int 
KfsClient::Rmdirs(const char *pathname)
{
    return mImpl->Rmdirs(pathname);
}

int 
KfsClient::Readdir(const char *pathname, std::vector<std::string> &result)
{
    return mImpl->Readdir(pathname, result);
}

int 
KfsClient::ReaddirPlus(const char *pathname, std::vector<KfsFileAttr> &result)
{
    return mImpl->ReaddirPlus(pathname, result);
}

int 
KfsClient::GetDirSummary(const char *pathname, uint64_t &numFiles, uint64_t &numBytes)
{
    return mImpl->GetDirSummary(pathname, numFiles, numBytes);
}

int 
KfsClient::Stat(const char *pathname, struct stat &result, bool computeFilesize)
{
    return mImpl->Stat(pathname, result, computeFilesize);
}

bool 
KfsClient::Exists(const char *pathname)
{
    return mImpl->Exists(pathname);
}

bool 
KfsClient::IsFile(const char *pathname)
{
    return mImpl->IsFile(pathname);
}

bool 
KfsClient::IsDirectory(const char *pathname)
{
    return mImpl->IsDirectory(pathname);
}

int
KfsClient::EnumerateBlocks(const char *pathname)
{
    return mImpl->EnumerateBlocks(pathname);
}

bool
KfsClient::VerifyDataChecksums(const char *pathname, const vector<uint32_t> &checksums)
{
    return mImpl->VerifyDataChecksums(pathname, checksums);
}

bool
KfsClient::VerifyDataChecksums(int fd, off_t offset, const char *buf, size_t numBytes)
{
    return mImpl->VerifyDataChecksums(fd, offset, buf, numBytes);
}

int 
KfsClient::Create(const char *pathname, int numReplicas, bool exclusive)
{
    return mImpl->Create(pathname, numReplicas, exclusive);
}

int 
KfsClient::Remove(const char *pathname)
{
    return mImpl->Remove(pathname);
}

int 
KfsClient::Rename(const char *oldpath, const char *newpath, bool overwrite)
{
    return mImpl->Rename(oldpath, newpath, overwrite);
}

int 
KfsClient::Open(const char *pathname, int openFlags, int numReplicas)
{
    return mImpl->Open(pathname, openFlags, numReplicas);
}

int 
KfsClient::Fileno(const char *pathname)
{
    return mImpl->Fileno(pathname);
}

int 
KfsClient::Close(int fd)
{
    return mImpl->Close(fd);
}

ssize_t 
KfsClient::Read(int fd, char *buf, size_t numBytes)
{
    return mImpl->Read(fd, buf, numBytes);
}

ssize_t 
KfsClient::Write(int fd, const char *buf, size_t numBytes)
{
    return mImpl->Write(fd, buf, numBytes);
}

int 
KfsClient::Sync(int fd)
{
    return mImpl->Sync(fd);
}

off_t 
KfsClient::Seek(int fd, off_t offset, int whence)
{
    return mImpl->Seek(fd, offset, whence);
}

off_t 
KfsClient::Seek(int fd, off_t offset)
{
    return mImpl->Seek(fd, offset, SEEK_SET);
}

off_t 
KfsClient::Tell(int fd)
{
    return mImpl->Tell(fd);
}

int 
KfsClient::Truncate(int fd, off_t offset)
{
    return mImpl->Truncate(fd, offset);
}

int 
KfsClient::GetDataLocation(const char *pathname, off_t start, size_t len,
                           std::vector< std::vector <std::string> > &locations)
{
    return mImpl->GetDataLocation(pathname, start, len, locations);
}

int 
KfsClient::GetDataLocation(int fd, off_t start, size_t len,
                           std::vector< std::vector <std::string> > &locations)
{
    return mImpl->GetDataLocation(fd, start, len, locations);
}

int16_t 
KfsClient::GetReplicationFactor(const char *pathname)
{
    return mImpl->GetReplicationFactor(pathname);
}

int16_t 
KfsClient::SetReplicationFactor(const char *pathname, int16_t numReplicas)
{
    return mImpl->SetReplicationFactor(pathname, numReplicas);
}

ServerLocation
KfsClient::GetMetaserverLocation() const
{
    return mImpl->GetMetaserverLocation();
}

//
// Now, the real work is done by the impl object....
//

KfsClientImpl::KfsClientImpl()
{
    pthread_mutexattr_t mutexAttr;
    int rval;
    const int hostnamelen = 256;
    char hostname[hostnamelen];

    if (gethostname(hostname, hostnamelen)) {
        perror("gethostname: ");
        exit(-1);
    }

    mHostname = hostname;

    // store the entry for "/"
    int UNUSED_ATTR rootfte = ClaimFileTableEntry(KFS::ROOTFID, "/", "/");
    assert(rootfte == 0);
    mFileTable[0]->fattr.fileId = KFS::ROOTFID;
    mFileTable[0]->fattr.isDirectory = true;

    mCwd = "/";
    mIsInitialized = false;
    mCmdSeqNum = 0;

    // Setup the mutex to allow recursive locking calls.  This
    // simplifies things when a public method (eg., read) in KFS client calls
    // another public method (eg., seek) and both need to take the lock
    rval = pthread_mutexattr_init(&mutexAttr);
    assert(rval == 0);
    rval = pthread_mutexattr_settype(&mutexAttr, PTHREAD_MUTEX_RECURSIVE);
    assert(rval == 0);
    rval = pthread_mutex_init(&mMutex, &mutexAttr);
    assert(rval == 0);

    // whenever a socket goes kaput, don't crash the app
    signal(SIGPIPE, SIG_IGN);

    // for random # generation, seed it
    srand(getpid());
}

int KfsClientImpl::Init(const string metaServerHost, int metaServerPort)
{
    // Initialize the logger
    MsgLogger::Init(NULL);

    MsgLogger::SetLevel(log4cpp::Priority::INFO);
    //MsgLogger::SetLevel(log4cpp::Priority::DEBUG);

    mRandSeed = time(NULL);

    mMetaServerLoc.hostname = metaServerHost;
    mMetaServerLoc.port = metaServerPort;

    KFS_LOG_VA_DEBUG("Connecting to metaserver at: %s:%d",
                     metaServerHost.c_str(), metaServerPort);

    if (!mMetaServerLoc.IsValid()) {
	mIsInitialized = false;
        KFS_LOG_VA_ERROR("Unable to connect to metaserver at: %s:%d",
                         metaServerHost.c_str(), metaServerPort);
	return -1;
    }

    if (!ConnectToMetaServer()) {
	mIsInitialized = false;
        KFS_LOG_VA_ERROR("Unable to connect to metaserver at: %s:%d",
                         metaServerHost.c_str(), metaServerPort);
	return -1;
    }


    // setup the telemetry stuff...
    struct ip_mreq imreq;
    string srvIp = "10.2.0.10";
    // string srvIp = "10.10.1.61";
    int srvPort = 12000;
    int multicastPort = 13000;

    imreq.imr_multiaddr.s_addr = inet_addr("172.16.17.32");
    imreq.imr_interface.s_addr = INADDR_ANY; // use DEFAULT interface
    
    // will setup this for release
    // mTelemetryReporter.Init(imreq, multicastPort, srvIp, srvPort);

    mIsInitialized = true;
    return 0;
}

bool
KfsClientImpl::ConnectToMetaServer()
{
    return mMetaServerSock.Connect(mMetaServerLoc) >= 0;
}


/// A notion of "cwd" in KFS.
///
int
KfsClientImpl::Cd(const char *pathname)
{
    MutexLock l(&mMutex);

    struct stat s;
    string path = build_path(mCwd, pathname);
    int status = Stat(path.c_str(), s);

    if (status < 0) {
	KFS_LOG_VA_DEBUG("Non-existent path: %s", pathname);
	return -ENOENT;
    }

    if (!S_ISDIR(s.st_mode)) {
	KFS_LOG_VA_DEBUG("Non-existent dir: %s", pathname);
	return -ENOTDIR;
    }

    // strip the trailing '/'
    string::size_type pathlen = path.size();
    string::size_type rslash = path.rfind('/');
    if (rslash + 1 == pathlen) {
        // path looks like: /.../; so, get rid of the last '/'
        path.erase(rslash);
    }

    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;
}

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

    struct stat statInfo;

    if (Stat(pathname, statInfo, false) != 0)
	return false;
    
    return S_ISREG(statInfo.st_mode);
}

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

    struct stat statInfo;

    if (Stat(pathname, statInfo, false) != 0)
	return false;
    
    return S_ISDIR(statInfo.st_mode);
}

int
KfsClientImpl::LookupAttr(kfsFileId_t parentFid, const char *filename,
	              KfsFileAttr &result, bool computeFilesize)
{
    MutexLock l(&mMutex);

    if (parentFid < 0)
	return -EINVAL;
    
    int fte = LookupFileTableEntry(parentFid, filename);
    LookupOp op(nextSeq(), parentFid, filename);

    if (fte >= 0) {
	result = mFileTable[fte]->fattr;
    } else {
        (void)DoMetaOpWithRetry(&op);
        if (op.status < 0)
            return op.status;

        result = op.fattr;
    }
    if ((!result.isDirectory) && computeFilesize) {
        if (result.fileSize < 0) {
            result.fileSize = ComputeFilesize(result.fileId);
        }
    }
    else {
        result.fileSize = -1;
    }

    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.fileSize;
        return 0;
    }

    // cache the entry if possible
    fte = AllocFileTableEntry(parentFid, filename, "");
    if (fte < 0)		
	return op.status;
    
    mFileTable[fte]->fattr = op.fattr;
    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.fileSize;

    return op.status;
}

int
KfsClientImpl::Create(const char *pathname, int numReplicas, bool exclusive)
{
    MutexLock l(&mMutex);

    kfsFileId_t parentFid;
    string filename;
    int res = GetPathComponents(pathname, &parentFid, filename);
    if (res < 0) {
	KFS_LOG_VA_DEBUG("status %d for pathname %s", res, pathname);
	return res;
    }

    if (filename.size() >= MAX_FILENAME_LEN)
	return -ENAMETOOLONG;

    CreateOp op(nextSeq(), parentFid, filename.c_str(), numReplicas, exclusive);
    (void)DoMetaOpWithRetry(&op);
    if (op.status < 0) {
	KFS_LOG_VA_DEBUG("status %d from create RPC", op.status);
	return op.status;
    }

    // Everything is good now...
    int fte = ClaimFileTableEntry(parentFid, filename.c_str(), pathname);
    if (fte < 0) {	// XXX Too many open files
	KFS_LOG_VA_DEBUG("status %d from ClaimFileTableEntry", fte);
	return fte;
    }

    FileAttr *fa = FdAttr(fte);
    fa->fileId = op.fileId;
    fa->Init(false);	// is an ordinary file

    FdInfo(fte)->openMode = O_RDWR;

    return fte;
}

int
KfsClientImpl::Remove(const char *pathname)
{
    MutexLock l(&mMutex);

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

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

    RemoveOp op(nextSeq(), parentFid, filename.c_str());
    (void)DoMetaOpWithRetry(&op);
    return op.status;
}

int
KfsClientImpl::Rename(const char *oldpath, const char *newpath, bool overwrite)
{
    MutexLock l(&mMutex);

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

    string absNewpath = build_path(mCwd, newpath);
    RenameOp op(nextSeq(), parentFid, oldfilename.c_str(),
		    absNewpath.c_str(), overwrite);
    (void)DoMetaOpWithRetry(&op);

    KFS_LOG_VA_DEBUG("Status of renaming %s -> %s is: %d", 
                     oldpath, newpath, op.status);

    // update the path cache
    if (op.status == 0) {
        int fte = LookupFileTableEntry(parentFid, oldfilename.c_str());
        if (fte > 0) {
            string oldn = string(oldpath);
            NameToFdMapIter iter = mPathCache.find(oldn);
        
            if (iter != mPathCache.end())
                mPathCache.erase(iter);

            mPathCache[absNewpath] = fte;
            mFileTable[fte]->pathname = absNewpath;
        }
    }

    return op.status;
}

int
KfsClientImpl::Fileno(const char *pathname)
{
    kfsFileId_t parentFid;
    string filename;
    int res = GetPathComponents(pathname, &parentFid, filename);
    if (res < 0)
	return res;

    return LookupFileTableEntry(parentFid, filename.c_str());
}

int
KfsClientImpl::Open(const char *pathname, int openMode, int numReplicas)
{
    MutexLock l(&mMutex);

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

    if (filename.size() >= MAX_FILENAME_LEN)
	return -ENAMETOOLONG;

    LookupOp op(nextSeq(), parentFid, filename.c_str());
    (void)DoMetaOpWithRetry(&op);

    if (op.status < 0) {
	if (openMode & O_CREAT) {
	    // file doesn't exist.  Create it
	    return Create(pathname, numReplicas, openMode & O_EXCL);
	}
	return op.status;
    } else {
        // file exists; now fail open if: O_CREAT | O_EXCL
        if ((openMode & (O_CREAT|O_EXCL)) == (O_CREAT|O_EXCL))
            return -EEXIST;
    }

    int fte = AllocFileTableEntry(parentFid, filename.c_str(), pathname);
    if (fte < 0)		// Too many open files
	return fte;

    // O_RDONLY is 0 and we use 0 to tell if the entry isn't used
    if ((openMode & O_RDWR) || (openMode & O_RDONLY))
	mFileTable[fte]->openMode = O_RDWR;
    else if (openMode & O_WRONLY)
	mFileTable[fte]->openMode = O_WRONLY;
    else
        // in this mode, we open the file to cache the attributes
        mFileTable[fte]->openMode = 0;

    // We got a path...get the fattr
    mFileTable[fte]->fattr = op.fattr;

    if (mFileTable[fte]->fattr.chunkCount > 0) {
	mFileTable[fte]->fattr.fileSize =
	    ComputeFilesize(op.fattr.fileId);
    }

    if (openMode & O_TRUNC)
	Truncate(fte, 0);

    if (openMode & O_APPEND)
	Seek(fte, 0, SEEK_END);

    return fte;
}

int
KfsClientImpl::Close(int fd)
{
    MutexLock l(&mMutex);
    int status = 0;

    if ((!valid_fd(fd)) || (mFileTable[fd] == NULL))
	return -EBADF;

    if (mFileTable[fd]->buffer.dirty) {
	status = FlushBuffer(fd);
    }
    KFS_LOG_VA_DEBUG("Closing filetable entry: %d", fd);
    ReleaseFileTableEntry(fd);

    return status;
}

int
KfsClientImpl::Sync(int fd)
{
    MutexLock l(&mMutex);

    if (!valid_fd(fd))
	return -EBADF;

    if (mFileTable[fd]->buffer.dirty) {
       int status = FlushBuffer(fd);
       if (status < 0)
	   return status;
    }
    return 0;
}

int
KfsClientImpl::Truncate(int fd, off_t offset)<commit_msg> -- Added a check for calling sync() on a closed fd. This was causing kfs-fuse  clients to occasionally fail.<commit_after>//---------------------------------------------------------- -*- Mode: C++ -*-
// $Id$ 
//
// Created 2006/04/18
// Author: <NAME>
//
// Copyright 2008 Quantcast Corp.
// Copyright 2006-2008 Kosmix Corp.
//
// This file is part of Kosmos File System (KFS).
//
// Licensed under the Apache License, Version 2.0
// (the "License"); you may not use this file except in compliance with
// the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
//
// \file KfsClient.cc
// \brief Kfs Client-library code.
//
//----------------------------------------------------------------------------

#include "KfsClient.h"
#include "KfsClientInt.h"

#include "common/config.h"
#include "common/properties.h"
#include "common/log.h"
#include "meta/kfstypes.h"
#include "libkfsIO/Checksum.h"
#include "Utils.h"

extern "C" {
#include <signal.h>
#include <stdlib.h>
}
#include <cerrno>
#include <iostream>
#include <string>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <boost/scoped_array.hpp>

using std::string;
using std::ostringstream;
using std::istringstream;
using std::min;
using std::max;
using std::map;
using std::vector;
using std::sort;
using std::transform;
using std::random_shuffle;

using std::cout;
using std::endl;

using namespace KFS;

const int CMD_BUF_SIZE = 1024;

// Set the default timeout for server I/O's to be 5 mins for now.
// This is intentionally large so that we can do stuff in gdb and not
// have the client timeout in the midst of a debug session.
struct timeval gDefaultTimeout = {300, 0};


namespace {
    Properties & theProps()
    {
        static Properties p;
        return p;
    }
}   

KfsClientFactory *
KFS::getKfsClientFactory()
{
    return KfsClientFactory::Instance();
}

KfsClientPtr
KfsClientFactory::GetClient(const char *propFile)
{
    bool verbose = false;
#ifdef DEBUG
    verbose = true;
#endif
    if (theProps().loadProperties(propFile, '=', verbose) != 0) {
        KfsClientPtr clnt;
	return clnt;
    }

    return GetClient(theProps().getValue("metaServer.name", ""),
                     theProps().getValue("metaServer.port", -1));

}

class MatchingServer {
    ServerLocation loc;
public:
    MatchingServer(const ServerLocation &l) : loc(l) { }
    bool operator()(KfsClientPtr &clnt) const {
        return clnt->GetMetaserverLocation() == loc;
    }
    bool operator()(const ServerLocation &other) const {
        return other == loc;
    }
};

KfsClientPtr
KfsClientFactory::GetClient(const std::string metaServerHost, int metaServerPort)
{
    vector<KfsClientPtr>::iterator iter;
    ServerLocation loc(metaServerHost, metaServerPort);

    iter = find_if(mClients.begin(), mClients.end(), MatchingServer(loc));
    if (iter != mClients.end())
        return *iter;

    KfsClientPtr clnt;

    clnt.reset(new KfsClient());

    clnt->Init(metaServerHost, metaServerPort);
    if (clnt->IsInitialized())
        mClients.push_back(clnt);
    else
        clnt.reset();

    return clnt;
}


KfsClient::KfsClient()
{
    mImpl = new KfsClientImpl();
}

KfsClient::~KfsClient()
{
    delete mImpl;
}

void
KfsClient::SetLogLevel(string logLevel)
{
    if (logLevel == "DEBUG")
        MsgLogger::SetLevel(log4cpp::Priority::DEBUG);
    else if (logLevel == "INFO")
        MsgLogger::SetLevel(log4cpp::Priority::INFO);
}

int 
KfsClient::Init(const std::string metaServerHost, int metaServerPort)
{
    return mImpl->Init(metaServerHost, metaServerPort);
}

bool 
KfsClient::IsInitialized()
{
    return mImpl->IsInitialized();
}

int
KfsClient::Cd(const char *pathname)
{
    return mImpl->Cd(pathname);
}

string
KfsClient::GetCwd()
{
    return mImpl->GetCwd();
}

int
KfsClient::Mkdirs(const char *pathname)
{
    return mImpl->Mkdirs(pathname);
}

int 
KfsClient::Mkdir(const char *pathname)
{
    return mImpl->Mkdir(pathname);
}

int 
KfsClient::Rmdir(const char *pathname)
{
    return mImpl->Rmdir(pathname);
}

int 
KfsClient::Rmdirs(const char *pathname)
{
    return mImpl->Rmdirs(pathname);
}

int 
KfsClient::Readdir(const char *pathname, std::vector<std::string> &result)
{
    return mImpl->Readdir(pathname, result);
}

int 
KfsClient::ReaddirPlus(const char *pathname, std::vector<KfsFileAttr> &result)
{
    return mImpl->ReaddirPlus(pathname, result);
}

int 
KfsClient::GetDirSummary(const char *pathname, uint64_t &numFiles, uint64_t &numBytes)
{
    return mImpl->GetDirSummary(pathname, numFiles, numBytes);
}

int 
KfsClient::Stat(const char *pathname, struct stat &result, bool computeFilesize)
{
    return mImpl->Stat(pathname, result, computeFilesize);
}

bool 
KfsClient::Exists(const char *pathname)
{
    return mImpl->Exists(pathname);
}

bool 
KfsClient::IsFile(const char *pathname)
{
    return mImpl->IsFile(pathname);
}

bool 
KfsClient::IsDirectory(const char *pathname)
{
    return mImpl->IsDirectory(pathname);
}

int
KfsClient::EnumerateBlocks(const char *pathname)
{
    return mImpl->EnumerateBlocks(pathname);
}

bool
KfsClient::VerifyDataChecksums(const char *pathname, const vector<uint32_t> &checksums)
{
    return mImpl->VerifyDataChecksums(pathname, checksums);
}

bool
KfsClient::VerifyDataChecksums(int fd, off_t offset, const char *buf, size_t numBytes)
{
    return mImpl->VerifyDataChecksums(fd, offset, buf, numBytes);
}

int 
KfsClient::Create(const char *pathname, int numReplicas, bool exclusive)
{
    return mImpl->Create(pathname, numReplicas, exclusive);
}

int 
KfsClient::Remove(const char *pathname)
{
    return mImpl->Remove(pathname);
}

int 
KfsClient::Rename(const char *oldpath, const char *newpath, bool overwrite)
{
    return mImpl->Rename(oldpath, newpath, overwrite);
}

int 
KfsClient::Open(const char *pathname, int openFlags, int numReplicas)
{
    return mImpl->Open(pathname, openFlags, numReplicas);
}

int 
KfsClient::Fileno(const char *pathname)
{
    return mImpl->Fileno(pathname);
}

int 
KfsClient::Close(int fd)
{
    return mImpl->Close(fd);
}

ssize_t 
KfsClient::Read(int fd, char *buf, size_t numBytes)
{
    return mImpl->Read(fd, buf, numBytes);
}

ssize_t 
KfsClient::Write(int fd, const char *buf, size_t numBytes)
{
    return mImpl->Write(fd, buf, numBytes);
}

int 
KfsClient::Sync(int fd)
{
    return mImpl->Sync(fd);
}

off_t 
KfsClient::Seek(int fd, off_t offset, int whence)
{
    return mImpl->Seek(fd, offset, whence);
}

off_t 
KfsClient::Seek(int fd, off_t offset)
{
    return mImpl->Seek(fd, offset, SEEK_SET);
}

off_t 
KfsClient::Tell(int fd)
{
    return mImpl->Tell(fd);
}

int 
KfsClient::Truncate(int fd, off_t offset)
{
    return mImpl->Truncate(fd, offset);
}

int 
KfsClient::GetDataLocation(const char *pathname, off_t start, size_t len,
                           std::vector< std::vector <std::string> > &locations)
{
    return mImpl->GetDataLocation(pathname, start, len, locations);
}

int 
KfsClient::GetDataLocation(int fd, off_t start, size_t len,
                           std::vector< std::vector <std::string> > &locations)
{
    return mImpl->GetDataLocation(fd, start, len, locations);
}

int16_t 
KfsClient::GetReplicationFactor(const char *pathname)
{
    return mImpl->GetReplicationFactor(pathname);
}

int16_t 
KfsClient::SetReplicationFactor(const char *pathname, int16_t numReplicas)
{
    return mImpl->SetReplicationFactor(pathname, numReplicas);
}

ServerLocation
KfsClient::GetMetaserverLocation() const
{
    return mImpl->GetMetaserverLocation();
}

//
// Now, the real work is done by the impl object....
//

KfsClientImpl::KfsClientImpl()
{
    pthread_mutexattr_t mutexAttr;
    int rval;
    const int hostnamelen = 256;
    char hostname[hostnamelen];

    if (gethostname(hostname, hostnamelen)) {
        perror("gethostname: ");
        exit(-1);
    }

    mHostname = hostname;

    // store the entry for "/"
    int UNUSED_ATTR rootfte = ClaimFileTableEntry(KFS::ROOTFID, "/", "/");
    assert(rootfte == 0);
    mFileTable[0]->fattr.fileId = KFS::ROOTFID;
    mFileTable[0]->fattr.isDirectory = true;

    mCwd = "/";
    mIsInitialized = false;
    mCmdSeqNum = 0;

    // Setup the mutex to allow recursive locking calls.  This
    // simplifies things when a public method (eg., read) in KFS client calls
    // another public method (eg., seek) and both need to take the lock
    rval = pthread_mutexattr_init(&mutexAttr);
    assert(rval == 0);
    rval = pthread_mutexattr_settype(&mutexAttr, PTHREAD_MUTEX_RECURSIVE);
    assert(rval == 0);
    rval = pthread_mutex_init(&mMutex, &mutexAttr);
    assert(rval == 0);

    // whenever a socket goes kaput, don't crash the app
    signal(SIGPIPE, SIG_IGN);

    // for random # generation, seed it
    srand(getpid());
}

int KfsClientImpl::Init(const string metaServerHost, int metaServerPort)
{
    // Initialize the logger
    MsgLogger::Init(NULL);

    MsgLogger::SetLevel(log4cpp::Priority::INFO);
    //MsgLogger::SetLevel(log4cpp::Priority::DEBUG);

    mRandSeed = time(NULL);

    mMetaServerLoc.hostname = metaServerHost;
    mMetaServerLoc.port = metaServerPort;

    KFS_LOG_VA_DEBUG("Connecting to metaserver at: %s:%d",
                     metaServerHost.c_str(), metaServerPort);

    if (!mMetaServerLoc.IsValid()) {
	mIsInitialized = false;
        KFS_LOG_VA_ERROR("Unable to connect to metaserver at: %s:%d",
                         metaServerHost.c_str(), metaServerPort);
	return -1;
    }

    if (!ConnectToMetaServer()) {
	mIsInitialized = false;
        KFS_LOG_VA_ERROR("Unable to connect to metaserver at: %s:%d",
                         metaServerHost.c_str(), metaServerPort);
	return -1;
    }


    // setup the telemetry stuff...
    struct ip_mreq imreq;
    string srvIp = "10.2.0.10";
    // int srvPort = 12000;
    //int multicastPort = 13000;

    imreq.imr_multiaddr.s_addr = inet_addr("172.16.17.32");
    imreq.imr_interface.s_addr = INADDR_ANY; // use DEFAULT interface
    
    // will setup this for release
    // mTelemetryReporter.Init(imreq, multicastPort, srvIp, srvPort);

    mIsInitialized = true;
    return 0;
}

bool
KfsClientImpl::ConnectToMetaServer()
{
    return mMetaServerSock.Connect(mMetaServerLoc) >= 0;
}


/// A notion of "cwd" in KFS.
///
int
KfsClientImpl::Cd(const char *pathname)
{
    MutexLock l(&mMutex);

    struct stat s;
    string path = build_path(mCwd, pathname);
    int status = Stat(path.c_str(), s);

    if (status < 0) {
	KFS_LOG_VA_DEBUG("Non-existent path: %s", pathname);
	return -ENOENT;
    }

    if (!S_ISDIR(s.st_mode)) {
	KFS_LOG_VA_DEBUG("Non-existent dir: %s", pathname);
	return -ENOTDIR;
    }

    // strip the trailing '/'
    string::size_type pathlen = path.size();
    string::size_type rslash = path.rfind('/');
    if (rslash + 1 == pathlen) {
        // path looks like: /.../; so, get rid of the last '/'
        path.erase(rslash);
    }

    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;
}

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

    struct stat statInfo;

    if (Stat(pathname, statInfo, false) != 0)
	return false;
    
    return S_ISREG(statInfo.st_mode);
}

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

    struct stat statInfo;

    if (Stat(pathname, statInfo, false) != 0)
	return false;
    
    return S_ISDIR(statInfo.st_mode);
}

int
KfsClientImpl::LookupAttr(kfsFileId_t parentFid, const char *filename,
	              KfsFileAttr &result, bool computeFilesize)
{
    MutexLock l(&mMutex);

    if (parentFid < 0)
	return -EINVAL;
    
    int fte = LookupFileTableEntry(parentFid, filename);
    LookupOp op(nextSeq(), parentFid, filename);

    if (fte >= 0) {
	result = mFileTable[fte]->fattr;
    } else {
        (void)DoMetaOpWithRetry(&op);
        if (op.status < 0)
            return op.status;

        result = op.fattr;
    }
    if ((!result.isDirectory) && computeFilesize) {
        if (result.fileSize < 0) {
            result.fileSize = ComputeFilesize(result.fileId);
        }
    }
    else {
        result.fileSize = -1;
    }

    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.fileSize;
        return 0;
    }

    // cache the entry if possible
    fte = AllocFileTableEntry(parentFid, filename, "");
    if (fte < 0)		
	return op.status;
    
    mFileTable[fte]->fattr = op.fattr;
    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.fileSize;

    return op.status;
}

int
KfsClientImpl::Create(const char *pathname, int numReplicas, bool exclusive)
{
    MutexLock l(&mMutex);

    kfsFileId_t parentFid;
    string filename;
    int res = GetPathComponents(pathname, &parentFid, filename);
    if (res < 0) {
	KFS_LOG_VA_DEBUG("status %d for pathname %s", res, pathname);
	return res;
    }

    if (filename.size() >= MAX_FILENAME_LEN)
	return -ENAMETOOLONG;

    CreateOp op(nextSeq(), parentFid, filename.c_str(), numReplicas, exclusive);
    (void)DoMetaOpWithRetry(&op);
    if (op.status < 0) {
	KFS_LOG_VA_DEBUG("status %d from create RPC", op.status);
	return op.status;
    }

    // Everything is good now...
    int fte = ClaimFileTableEntry(parentFid, filename.c_str(), pathname);
    if (fte < 0) {	// XXX Too many open files
	KFS_LOG_VA_DEBUG("status %d from ClaimFileTableEntry", fte);
	return fte;
    }

    FileAttr *fa = FdAttr(fte);
    fa->fileId = op.fileId;
    fa->Init(false);	// is an ordinary file

    FdInfo(fte)->openMode = O_RDWR;

    return fte;
}

int
KfsClientImpl::Remove(const char *pathname)
{
    MutexLock l(&mMutex);

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

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

    RemoveOp op(nextSeq(), parentFid, filename.c_str());
    (void)DoMetaOpWithRetry(&op);
    return op.status;
}

int
KfsClientImpl::Rename(const char *oldpath, const char *newpath, bool overwrite)
{
    MutexLock l(&mMutex);

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

    string absNewpath = build_path(mCwd, newpath);
    RenameOp op(nextSeq(), parentFid, oldfilename.c_str(),
		    absNewpath.c_str(), overwrite);
    (void)DoMetaOpWithRetry(&op);

    KFS_LOG_VA_DEBUG("Status of renaming %s -> %s is: %d", 
                     oldpath, newpath, op.status);

    // update the path cache
    if (op.status == 0) {
        int fte = LookupFileTableEntry(parentFid, oldfilename.c_str());
        if (fte > 0) {
            string oldn = string(oldpath);
            NameToFdMapIter iter = mPathCache.find(oldn);
        
            if (iter != mPathCache.end())
                mPathCache.erase(iter);

            mPathCache[absNewpath] = fte;
            mFileTable[fte]->pathname = absNewpath;
        }
    }

    return op.status;
}

int
KfsClientImpl::Fileno(const char *pathname)
{
    kfsFileId_t parentFid;
    string filename;
    int res = GetPathComponents(pathname, &parentFid, filename);
    if (res < 0)
	return res;

    return LookupFileTableEntry(parentFid, filename.c_str());
}

int
KfsClientImpl::Open(const char *pathname, int openMode, int numReplicas)
{
    MutexLock l(&mMutex);

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

    if (filename.size() >= MAX_FILENAME_LEN)
	return -ENAMETOOLONG;

    LookupOp op(nextSeq(), parentFid, filename.c_str());
    (void)DoMetaOpWithRetry(&op);

    if (op.status < 0) {
	if (openMode & O_CREAT) {
	    // file doesn't exist.  Create it
	    return Create(pathname, numReplicas, openMode & O_EXCL);
	}
	return op.status;
    } else {
        // file exists; now fail open if: O_CREAT | O_EXCL
        if ((openMode & (O_CREAT|O_EXCL)) == (O_CREAT|O_EXCL))
            return -EEXIST;
    }

    int fte = AllocFileTableEntry(parentFid, filename.c_str(), pathname);
    if (fte < 0)		// Too many open files
	return fte;

    // O_RDONLY is 0 and we use 0 to tell if the entry isn't used
    if ((openMode & O_RDWR) || (openMode & O_RDONLY))
	mFileTable[fte]->openMode = O_RDWR;
    else if (openMode & O_WRONLY)
	mFileTable[fte]->openMode = O_WRONLY;
    else
        // in this mode, we open the file to cache the attributes
        mFileTable[fte]->openMode = 0;

    // We got a path...get the fattr
    mFileTable[fte]->fattr = op.fattr;

    if (mFileTable[fte]->fattr.chunkCount > 0) {
	mFileTable[fte]->fattr.fileSize =
	    ComputeFilesize(op.fattr.fileId);
    }

    if (openMode & O_TRUNC)
	Truncate(fte, 0);

    if (openMode & O_APPEND)
	Seek(fte, 0, SEEK_END);

    return fte;
}

int
KfsClientImpl::Close(int fd)
{
    MutexLock l(&mMutex);
    int status = 0;

    if ((!valid_fd(fd)) || (mFileTable[fd] == NULL))
	return -EBADF;

    if (mFileTable[fd]->buffer.dirty) {
	status = FlushBuffer(fd);
    }
    KFS_LOG_VA_DEBUG("Closing filetable entry: %d", fd);
    ReleaseFileTableEntry(fd);

    return status;
}

int
KfsClientImpl::Sync(int fd)
{
    MutexLock l(&mMutex);

    if ((!valid_fd(fd)) || (mFileTable[fd] == NULL))
	return -EBADF;

    if (mFileTable[fd]->buffer.dirty) {
       int status = FlushBuffer(fd);
       if (status < 0)
	   return status;
    }
    return 0;
}

int
KfsClientImpl::Truncate(int fd, off_t offset)
{'fraction_non_alphanumeric': 0.1351683383873727, 'fraction_numerical': 0.006982349739522588, 'mean_word_length': 2.749839640795382, 'pattern_counts': {'":': 0, '<': 145, '<?xml version=': 0, '>': 249, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 5, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6866133', 'n_tokens_mistral': 22385, 'n_tokens_neox': 21598, 'n_words': 5450}