BESFileLockingCache.cc

Go to the documentation of this file.

// BESFileLockingCache.cc

// This file was originally part of bes, A C++ back-end server
// implementation framework for the OPeNDAP Data Access Protocol.
// Copied to libdap. This is used to cache responses built from
// functional CE expressions.

// Moved back to the BES. 6/11/13 jhrg

// Copyright (c) 2012 OPeNDAP, Inc
// Author: James Gallagher <jgallagher@opendap.org>
// Patrick West <pwest@ucar.edu> and Jose Garcia <jgarcia@ucar.edu>
//
// This library is free software; you can redistribute it and/or
// modify it under the terms of the GNU Lesser General Public
// License as published by the Free Software Foundation; either
// version 2.1 of the License, or (at your option) any later version.
//
// This library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
// Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public
// License along with this library; if not, write to the Free Software
// Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
//
// You can contact University Corporation for Atmospheric Research at
// 3080 Center Green Drive, Boulder, CO 80301

#include "config.h"

#include <sys/file.h>
#include <sys/stat.h>
#include <unistd.h>
#include <dirent.h>
#include <fcntl.h>

#ifdef HAVE_STDLIB_H
#include <stdlib.h>
#endif

#include <string>
#include <sstream>
#include <vector>
#include <cstring>
#include <cerrno>

//#define DODS_DEBUG

#include "BESInternalError.h"

#include "BESDebug.h"
#include "BESLog.h"

#include "BESFileLockingCache.h"

using namespace std;

// conversion factor
static const unsigned long long BYTES_PER_MEG = 1048576ULL;

// Max cache size in megs, so we can check the user input and warn.
// 2^64 / 2^20 == 2^44
static const unsigned long long MAX_CACHE_SIZE_IN_MEGABYTES = (1ULL << 44);

BESFileLockingCache::BESFileLockingCache(const string &cache_dir, const string &prefix, unsigned long long size) :
        d_cache_dir(cache_dir), d_prefix(prefix), d_max_cache_size_in_bytes(size)
{
    m_initialize_cache_info();
}

void BESFileLockingCache::initialize(const string &cache_dir, const string &prefix, unsigned long long size)
{
        d_cache_dir = cache_dir;
        d_prefix = prefix;
        d_max_cache_size_in_bytes = size;

    m_initialize_cache_info();
}

static inline string get_errno() {
        char *s_err = strerror(errno);
        if (s_err)
                return s_err;
        else
                return "Unknown error.";
}

// Build a lock of a certain type.
static inline struct flock *lock(int type) {
    static struct flock lock;
    lock.l_type = type;
    lock.l_whence = SEEK_SET;
    lock.l_start = 0;
    lock.l_len = 0;
    lock.l_pid = getpid();

    return &lock;
}

inline void BESFileLockingCache::m_record_descriptor(const string &file, int fd) {
    BESDEBUG("cache", "DAP Cache: recording descriptor: " << file << ", " << fd << endl);
    d_locks.insert(std::pair<string, int>(file, fd));
}

inline int BESFileLockingCache::m_get_descriptor(const string &file) {
        BESDEBUG("cache", "BESFileLockingCache::m_get_descriptor; d_locks size: " << d_locks.size() << endl);
    FilesAndLockDescriptors::iterator i = d_locks.find(file);
    if (i == d_locks.end())
        return -1;

    int fd = i->second;
    BESDEBUG("cache", "DAP Cache: getting descriptor: " << file << ", " << fd << endl);
    d_locks.erase(i);
    return fd;
}

static void unlock(int fd)
{
    if (fcntl(fd, F_SETLK, lock(F_UNLCK)) == -1) {
        throw BESInternalError( "An error occurred trying to unlock the file: " + get_errno(), __FILE__, __LINE__);
    }

    if (close(fd) == -1)
        throw BESInternalError("Could not close the (just) unlocked file.", __FILE__, __LINE__);
}

static bool getSharedLock(const string &file_name, int &ref_fd)
{
        BESDEBUG("cache", "getSharedLock: " << file_name <<endl);

    int fd;
    if ((fd = open(file_name.c_str(), O_RDONLY)) < 0) {
        switch (errno) {
        case ENOENT:
            return false;

        default:
            throw BESInternalError( get_errno(), __FILE__, __LINE__);
        }
    }

    struct flock *l = lock(F_RDLCK);
    if (fcntl(fd, F_SETLKW, l) == -1) {
        close(fd);
        ostringstream oss;
        oss << "cache process: " << l->l_pid << " triggered a locking error: " << get_errno();
        throw BESInternalError( oss.str(), __FILE__, __LINE__);
    }

    BESDEBUG("cache", "getSharedLock exit: " << file_name <<endl);

    // Success
    ref_fd = fd;
    return true;
}

bool BESFileLockingCache::getExclusiveLock(string file_name, int &ref_fd)
{
        BESDEBUG("cache", "BESFileLockingCache::getExclusiveLock() - " << file_name <<endl);

    int fd;
    if ((fd = open(file_name.c_str(), O_RDWR)) < 0) {
        switch (errno) {
        case ENOENT:
            return false;

        default:
            throw BESInternalError(get_errno(), __FILE__, __LINE__);
        }
    }

    struct flock *l = lock(F_WRLCK);
    if (fcntl(fd, F_SETLKW, l) == -1) {
        close(fd);
        ostringstream oss;
        oss << "cache process: " << l->l_pid << " triggered a locking error: " << get_errno();
        throw BESInternalError(oss.str(), __FILE__, __LINE__);
    }

    BESDEBUG("cache", "BESFileLockingCache::getExclusiveLock() -  exit: " << file_name <<endl);

    // Success
    ref_fd = fd;
    return true;
}

static bool getExclusiveLockNB(string file_name, int &ref_fd)
{
        BESDEBUG("cache", "getExclusiveLock_nonblocking: " << file_name <<endl);

    int fd;
    if ((fd = open(file_name.c_str(), O_RDWR)) < 0) {
        switch (errno) {
        case ENOENT:
            return false;

        default:
            throw BESInternalError(get_errno(), __FILE__, __LINE__);
        }
    }

    struct flock *l = lock(F_WRLCK);
    if (fcntl(fd, F_SETLK, l) == -1) {
        switch (errno) {
        case EAGAIN:
                BESDEBUG("cache", "getExclusiveLock_nonblocking exit (false): " << file_name << " by: " << l->l_pid << endl);
            close(fd);
            return false;

        default: {
            close(fd);
                ostringstream oss;
                oss << "cache process: " << l->l_pid << " triggered a locking error: " << get_errno();
                throw BESInternalError(oss.str(), __FILE__, __LINE__);
        }
        }
    }

    BESDEBUG("cache", "getExclusiveLock_nonblocking exit (true): " << file_name <<endl);

    // Success
    ref_fd = fd;
    return true;
}

static bool createLockedFile(string file_name, int &ref_fd)
{
        BESDEBUG("cache", "createLockedFile: " << file_name <<endl);

    int fd;
    if ((fd = open(file_name.c_str(), O_CREAT | O_EXCL | O_RDWR, 0666)) < 0) {
        switch (errno) {
        case EEXIST:
            return false;

        default:
            throw BESInternalError(get_errno(), __FILE__, __LINE__);
        }
    }

    struct flock *l = lock(F_WRLCK);
    if (fcntl(fd, F_SETLKW, l) == -1) {
        close(fd);
        ostringstream oss;
        oss << "cache process: " << l->l_pid << " triggered a locking error: " << get_errno();
        throw BESInternalError(oss.str(), __FILE__, __LINE__);
    }

    BESDEBUG("cache", "createLockedFile exit: " << file_name <<endl);

    // Success
    ref_fd = fd;
    return true;
}

void BESFileLockingCache::m_check_ctor_params()
{
    if (d_cache_dir.empty()) {
        string err = "The cache directory was not specified";
        throw BESInternalError(err, __FILE__, __LINE__);
    }

    struct stat buf;
    int statret = stat(d_cache_dir.c_str(), &buf);
    if (statret != 0 || !S_ISDIR(buf.st_mode)) {
        // Try to make the directory
        int status = mkdir(d_cache_dir.c_str(), 0775);
        if (status != 0) {
            string err = "The cache directory " + d_cache_dir + " does not exist or could not be created.";
            throw BESInternalError(err, __FILE__, __LINE__);
        }
    }

    if (d_prefix.empty()) {
        string err = "The cache file prefix was not specified, must not be empty";
        throw BESInternalError(err, __FILE__, __LINE__);
    }

    if (d_max_cache_size_in_bytes <= 0) {
        string err = "The cache size was not specified, must be greater than zero";
        throw BESInternalError(err, __FILE__, __LINE__);
    }

    BESDEBUG("cache", "DAP Cache: directory " << d_cache_dir << ", prefix " << d_prefix
            << ", max size " << d_max_cache_size_in_bytes << endl );
}

void BESFileLockingCache::m_initialize_cache_info()
{
    // The value set in configuration files, etc., is the size in megabytes. The private
    // variable holds the size in bytes (converted below).
    d_max_cache_size_in_bytes = min(d_max_cache_size_in_bytes, MAX_CACHE_SIZE_IN_MEGABYTES);
    d_max_cache_size_in_bytes *= BYTES_PER_MEG;
    d_target_size = d_max_cache_size_in_bytes * 0.8;

    m_check_ctor_params(); // Throws BESInternalError on error.

    d_cache_info = d_cache_dir + "/dap.cache.info";

    // See if we can create it. If so, that means it doesn't exist. So make it and
    // set the cache initial size to zero.
    if (createLockedFile(d_cache_info, d_cache_info_fd)) {
                // initialize the cache size to zero
                unsigned long long size = 0;
                if (write(d_cache_info_fd, &size, sizeof(unsigned long long)) != sizeof(unsigned long long))
                        throw BESInternalError("Could not write size info to the cache info file `"+d_cache_info+"`", __FILE__, __LINE__);

                // This leaves the d_cache_info_fd file descriptor open
                unlock_cache();
        }
        else {
                if ((d_cache_info_fd = open(d_cache_info.c_str(), O_RDWR)) == -1) {
                        throw BESInternalError(get_errno(), __FILE__, __LINE__);
                }
        }

    // cerr << endl << "***** BESFileLockingCache::m_initialize_cache_info() completed. d_cache_info_fd: " << d_cache_info_fd << endl;

    BESDEBUG("cache", "d_cache_info_fd: " << d_cache_info_fd << endl);
}

string BESFileLockingCache::get_cache_file_name(const string &src, bool mangle)
{
    string target = src;

    if (mangle) {
        if (target.at(0) == '/') {
            target = src.substr(1, target.length() - 1);
        }
        string::size_type slash = 0;
        while ((slash = target.find('/')) != string::npos) {
            target.replace(slash, 1, 1, BESFileLockingCache::DAP_CACHE_CHAR);
        }
        string::size_type last_dot = target.rfind('.');
        if (last_dot != string::npos) {
            target = target.substr(0, last_dot);
        }
    }
    BESDEBUG("cache", "  d_cache_dir: '" << d_cache_dir << "'" << endl);
    BESDEBUG("cache", "  d_prefix:    '" << d_prefix << "'" << endl);
    BESDEBUG("cache", "  target:      '" << target  << "'" << endl);

    return d_cache_dir + "/" + d_prefix + BESFileLockingCache::DAP_CACHE_CHAR + target;
}

bool BESFileLockingCache::get_read_lock(const string &target, int &fd)
{
        lock_cache_read();

    bool status = getSharedLock(target, fd);

    BESDEBUG("cache", "BESFileLockingCache: get_read_lock: " << target << " (status: " << status << ", fd: " << fd << ")" << endl);

    if (status)
        m_record_descriptor(target, fd);

    unlock_cache();

    return status;
}

bool BESFileLockingCache::create_and_lock(const string &target, int &fd)
{
        lock_cache_write();

    bool status = createLockedFile(target, fd);

    BESDEBUG("cache", "BESFileLockingCache: create_and_lock: " << target << " (status: " << status << ", fd: " << fd << ")" << endl);

    if (status)
        m_record_descriptor(target, fd);

    unlock_cache();

    return status;

}

void BESFileLockingCache::exclusive_to_shared_lock(int fd)
{
    struct flock lock;
    lock.l_type = F_RDLCK;
    lock.l_whence = SEEK_SET;
    lock.l_start = 0;
    lock.l_len = 0;
    lock.l_pid = getpid();

    if (fcntl(fd, F_SETLKW, &lock) == -1) {
        throw BESInternalError(get_errno(), __FILE__, __LINE__);
    }
}

void BESFileLockingCache::lock_cache_write()
{
        BESDEBUG("cache", "lock_cache - d_cache_info_fd: " << d_cache_info_fd << endl);

    if (fcntl(d_cache_info_fd, F_SETLKW, lock(F_WRLCK)) == -1) {
        throw BESInternalError("An error occurred trying to lock the cache-control file" + get_errno(), __FILE__, __LINE__);
    }
}

void BESFileLockingCache::lock_cache_read()
{
        BESDEBUG("cache", "lock_cache - d_cache_info_fd: " << d_cache_info_fd << endl);

    if (fcntl(d_cache_info_fd, F_SETLKW, lock(F_RDLCK)) == -1) {
        throw BESInternalError("An error occurred trying to lock the cache-control file" + get_errno(), __FILE__, __LINE__);
    }
}

void BESFileLockingCache::unlock_cache()
{
        BESDEBUG("cache", "DAP Cache: unlock: cache_info (fd: " << d_cache_info_fd << ")" << endl);

    if (fcntl(d_cache_info_fd, F_SETLK, lock(F_UNLCK)) == -1) {
        throw BESInternalError("An error occurred trying to unlock the cache-control file" + get_errno(), __FILE__, __LINE__);
    }
}

void BESFileLockingCache::unlock_and_close(const string &file_name)
{
        BESDEBUG("cache", "DAP Cache: unlock file: " << file_name << endl);

    int fd = m_get_descriptor(file_name);       // returns -1 when no more files desp. remain
    while (fd != -1) {
        unlock(fd);
        fd = m_get_descriptor(file_name);
    }
}

#if 0
// I removed this because it will unlock() the file descriptor without
// removing it from the list of registered fds. This can lead to errors
// when the other version is called because a given descriptor appears
// more than once in the list of descriptors or a fd in the list is no
// longer open. jhrg 8/13/14

void BESFileLockingCache::unlock_and_close(int fd)
{
        BESDEBUG("cache", "DAP Cache: unlock fd: " << fd << endl);

    unlock(fd);

    BESDEBUG("cache", "DAP Cache: unlock " << fd << " Success" << endl);
}
#endif

unsigned long long BESFileLockingCache::update_cache_info(const string &target)
{
        unsigned long long current_size;
        try {
                lock_cache_write();

                if (lseek(d_cache_info_fd, 0, SEEK_SET) == -1)
                        throw BESInternalError("Could not rewind to front of cache info file.", __FILE__, __LINE__);

                // read the size from the cache info file
                if (read(d_cache_info_fd, &current_size, sizeof(unsigned long long)) != sizeof(unsigned long long))
                        throw BESInternalError("Could not get read size info from the cache info file!", __FILE__, __LINE__);

                struct stat buf;
                int statret = stat(target.c_str(), &buf);
                if (statret == 0)
                        current_size += buf.st_size;
                else
                        throw BESInternalError("Could not read the size of the new file: " + target + " : " + get_errno(), __FILE__, __LINE__);

                BESDEBUG("cache", "DAP Cache: cache size updated to: " << current_size << endl);

                if (lseek(d_cache_info_fd, 0, SEEK_SET) == -1)
                        throw BESInternalError("Could not rewind to front of cache info file.", __FILE__, __LINE__);

                if(write(d_cache_info_fd, &current_size, sizeof(unsigned long long)) != sizeof(unsigned long long))
                        throw BESInternalError("Could not write size info from the cache info file!", __FILE__, __LINE__);

                unlock_cache();
        }
        catch (...) {
                unlock_cache();
                throw;
        }

        return current_size;
}

bool BESFileLockingCache::cache_too_big(unsigned long long current_size) const
{
    return current_size > d_max_cache_size_in_bytes;
}

unsigned long long BESFileLockingCache::get_cache_size()
{
        unsigned long long current_size;
        try {
                lock_cache_read();

                if (lseek(d_cache_info_fd, 0, SEEK_SET) == -1)
                throw BESInternalError("Could not rewind to front of cache info file.", __FILE__, __LINE__);
                // read the size from the cache info file
                if(read(d_cache_info_fd, &current_size, sizeof(unsigned long long)) != sizeof(unsigned long long))
                    throw BESInternalError("Could not get read size info from the cache info file!", __FILE__, __LINE__);

                unlock_cache();
        }
        catch(...) {
                unlock_cache();
                throw;
        }

        return current_size;
}


static bool entry_op(cache_entry &e1, cache_entry &e2)
{
    return e1.time < e2.time;
}

unsigned long long BESFileLockingCache::m_collect_cache_dir_info(CacheFiles &contents)
{
    DIR *dip = opendir(d_cache_dir.c_str());
    if (!dip)
        throw BESInternalError("Unable to open cache directory " + d_cache_dir, __FILE__, __LINE__);

    struct dirent *dit;
    vector<string> files;
    // go through the cache directory and collect all of the files that
    // start with the matching prefix
    while ((dit = readdir(dip)) != NULL) {
        string dirEntry = dit->d_name;
        if (dirEntry.compare(0, d_prefix.length(), d_prefix) == 0) {
            files.push_back(d_cache_dir + "/" + dirEntry);
        }
    }

    closedir(dip);

    unsigned long long current_size = 0;
    struct stat buf;
    for (vector<string>::iterator file = files.begin(); file != files.end(); ++file) {
        if (stat(file->c_str(), &buf) == 0) {
            current_size += buf.st_size;
            cache_entry entry;
            entry.name = *file;
            entry.size = buf.st_size;
            entry.time = buf.st_atime;
            // Sanity check; Removed after initial testing since some files might be zero bytes
#if 0
            if (entry.size == 0)
                throw BESInternalError("Zero-byte file found in cache. " + *file, __FILE__, __LINE__);
#endif
            contents.push_back(entry);
        }
    }

    // Sort so smaller (older) times are first.
    contents.sort(entry_op);

    return current_size;
}

void BESFileLockingCache::update_and_purge(const string &new_file)
{
        BESDEBUG("cache", "purge - starting the purge" << endl);

    try {
        lock_cache_write();

        CacheFiles contents;
        unsigned long long computed_size = m_collect_cache_dir_info(contents);
#if 0
        if (BESISDEBUG( "cache_contents" )) {
                BESDEBUG("cache", "BEFORE Purge " << computed_size/BYTES_PER_MEG << endl );
            CacheFiles::iterator ti = contents.begin();
            CacheFiles::iterator te = contents.end();
            for (; ti != te; ti++) {
                BESDEBUG("cache", (*ti).time << ": " << (*ti).name << ": size " << (*ti).size/BYTES_PER_MEG << endl );
            }
        }
#endif
        BESDEBUG("cache", "purge - current and target size (in MB) " << computed_size/BYTES_PER_MEG  << ", " << d_target_size/BYTES_PER_MEG << endl );

        // This deletes files and updates computed_size
        if (cache_too_big(computed_size)) {

            // d_target_size is 80% of the maximum cache size.
            // Grab the first which is the oldest in terms of access time.
            CacheFiles::iterator i = contents.begin();
            while (i != contents.end() && computed_size > d_target_size) {
                // Grab an exclusive lock but do not block - if another process has the file locked
                // just move on to the next file. Also test to see if the current file is the file
                // this process just added to the cache - don't purge that!
                int cfile_fd;
                if (i->name != new_file && getExclusiveLockNB(i->name, cfile_fd)) {
                        BESDEBUG("cache", "purge: " << i->name << " removed." << endl );

                    if (unlink(i->name.c_str()) != 0)
                        throw BESInternalError("Unable to purge the file " + i->name + " from the cache: " + get_errno(), __FILE__, __LINE__);

                    unlock(cfile_fd);
                    computed_size -= i->size;
                }
                ++i;

                BESDEBUG("cache", "purge - current and target size (in MB) " << computed_size/BYTES_PER_MEG << ", " << d_target_size/BYTES_PER_MEG << endl );
            }

        }

        if (lseek(d_cache_info_fd, 0, SEEK_SET) == -1)
            throw BESInternalError("Could not rewind to front of cache info file.", __FILE__, __LINE__);

        if(write(d_cache_info_fd, &computed_size, sizeof(unsigned long long)) != sizeof(unsigned long long))
            throw BESInternalError("Could not write size info to the cache info file!", __FILE__, __LINE__);
#if 0
        if (BESISDEBUG( "cache_contents" )) {
            contents.clear();
            computed_size = m_collect_cache_dir_info(contents);
            BESDEBUG("cache", "AFTER Purge " << computed_size/BYTES_PER_MEG << endl );
            CacheFiles::iterator ti = contents.begin();
            CacheFiles::iterator te = contents.end();
            for (; ti != te; ti++) {
                BESDEBUG("cache", (*ti).time << ": " << (*ti).name << ": size " << (*ti).size/BYTES_PER_MEG << endl );
            }
        }
#endif
        unlock_cache();
    }
    catch(...) {
        unlock_cache();
        throw;
    }
}

void BESFileLockingCache::purge_file(const string &file)
{
        BESDEBUG("cache", "BESFileLockingCache::purge_file() - starting the purge" << endl);

    try {
        lock_cache_write();

        // Grab an exclusive lock on the file
        int cfile_fd;
        if (getExclusiveLock(file, cfile_fd)) {
            // Get the file's size
            unsigned long long size = 0;
            struct stat buf;
            if (stat(file.c_str(), &buf) == 0) {
                size = buf.st_size;
            }

            BESDEBUG("cache", "BESFileLockingCache::purge_file() - " << file << " removed." << endl );

            if (unlink(file.c_str()) != 0)
                throw BESInternalError("Unable to purge the file " + file + " from the cache: " + get_errno(), __FILE__, __LINE__);

            unlock(cfile_fd);

            unsigned long long cache_size = get_cache_size() - size;

            if (lseek(d_cache_info_fd, 0, SEEK_SET) == -1)
                throw BESInternalError("Could not rewind to front of cache info file.", __FILE__, __LINE__);

            if (write(d_cache_info_fd, &cache_size, sizeof(unsigned long long)) != sizeof(unsigned long long))
                throw BESInternalError("Could not write size info to the cache info file!", __FILE__, __LINE__);
        }

        unlock_cache();
    }
    catch (...) {
        unlock_cache();
        throw;
    }
}


const string BESFileLockingCache::getCacheFilePrefix(){
        return d_prefix;
}

const string BESFileLockingCache::getCacheDirectory(){
        return d_cache_dir;
}



void BESFileLockingCache::dump(ostream &strm) const
{
    strm << BESIndent::LMarg << "BESFileLockingCache::dump - (" << (void *) this << ")" << endl;
    BESIndent::Indent();
    strm << BESIndent::LMarg << "cache dir: " << d_cache_dir << endl;
    strm << BESIndent::LMarg << "prefix: " << d_prefix << endl;
    strm << BESIndent::LMarg << "size (bytes): " << d_max_cache_size_in_bytes << endl;
    BESIndent::UnIndent();
}