DAPCache3.cc

// DAPCache3.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.

// 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>

#include "DAPCache3.h"

//#define DODS_DEBUG

#include "InternalErr.h"
#include "DapIndent.h"
#include "debug.h"

#if 0
#include "BESSyntaxUserError.h"
#include "BESInternalError.h"

#include "TheBESKeys.h"
#include "BESDebug.h"
#include "BESLog.h"
#endif
using namespace std;
using namespace libdap;

namespace libdap {

// 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);

DAPCache3 *DAPCache3::d_instance = 0;


DAPCache3::DAPCache3(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 DAPCache3::delete_instance() {
    DBG(cerr << "DAPCache3::delete_instance() - Deleting singleton DAPCache3 instance." << endl);
    delete d_instance;
    d_instance = 0;
}

#if 0
// The BESCache3 code is a singleton that assumes it's running in the absence of threads but that
// the cache is shared by several processes, each of which have their own instance of BESCache3.
BESCache3 *
BESCache3::get_instance(BESKeys *keys, const string &cache_dir_key, const string &prefix_key, const string &size_key)
{
    if (d_instance == 0)
        d_instance = new BESCache3(keys, cache_dir_key, prefix_key, size_key);

    return d_instance;
}
#endif

DAPCache3 *
DAPCache3::get_instance(const string &cache_dir, const string &prefix, unsigned long long size)
{
    if (d_instance == 0){
        d_instance = new DAPCache3(cache_dir, prefix, size);
#if HAVE_ATEXIT
        atexit(delete_instance);
#endif
    }
    return d_instance;
}

DAPCache3 *
DAPCache3::get_instance()
{
    if (d_instance == 0)
        throw InternalErr(__FILE__, __LINE__, "Tried to get the DAPCache3 instance, but it hasn't been created yet");

    return d_instance;
}

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 DAPCache3::m_record_descriptor(const string &file, int fd) {
    DBG(cerr << "DAP Cache: recording descriptor: " << file << ", " << fd << endl);
    d_locks.insert(std::pair<string, int>(file, fd));
}

inline int DAPCache3::m_get_descriptor(const string &file) {
    FilesAndLockDescriptors::iterator i = d_locks.find(file);
    int fd = i->second;
    DBG(cerr << "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 InternalErr(__FILE__, __LINE__, "An error occurred trying to unlock the file" + get_errno());
    }

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

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

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

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

    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 InternalErr(__FILE__, __LINE__, oss.str());
    }

    DBG(cerr << "getSharedLock exit: " << file_name <<endl);

    // Success
    ref_fd = fd;
    return true;
}

static bool getExclusiveLock(string file_name, int &ref_fd)
{
    DBG(cerr << "getExclusiveLock: " << file_name <<endl);

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

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

    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 InternalErr(__FILE__, __LINE__, oss.str());
    }

    DBG(cerr << "getExclusiveLock exit: " << file_name <<endl);

    // Success
    ref_fd = fd;
    return true;
}

static bool getExclusiveLockNB(string file_name, int &ref_fd)
{
    DBG(cerr << "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 InternalErr(__FILE__, __LINE__, get_errno());
        }
    }

    struct flock *l = lock(F_WRLCK);
    if (fcntl(fd, F_SETLK, l) == -1) {
        switch (errno) {
        case EAGAIN:
            DBG(cerr << "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 InternalErr(__FILE__, __LINE__, oss.str());
        }
        }
    }

    DBG(cerr << "getExclusiveLock_nonblocking exit (true): " << file_name <<endl);

    // Success
    ref_fd = fd;
    return true;
}

static bool createLockedFile(string file_name, int &ref_fd)
{
    DBG(cerr << "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 InternalErr(__FILE__, __LINE__, get_errno());
        }
    }

    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 InternalErr(__FILE__, __LINE__, oss.str());
    }

    DBG(cerr << "createLockedFile exit: " << file_name <<endl);

    // Success
    ref_fd = fd;
    return true;
}

void DAPCache3::m_check_ctor_params()
{
    if (d_cache_dir.empty()) {
        string err = "The cache directory was not specified, must be non-empty";
        throw InternalErr(__FILE__, __LINE__, err);
    }

    // TODO New feature: Makes the directory f it does not exist
    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 InternalErr(__FILE__, __LINE__, err);
        }
    }

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

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

    // If the user specifies a cache that is too large,
    // it is a user exception and we should tell them.
    if (d_max_cache_size_in_bytes > MAX_CACHE_SIZE_IN_MEGABYTES) {
        std::ostringstream msg;
        msg << "The specified cache size was larger than the max cache size of: " << MAX_CACHE_SIZE_IN_MEGABYTES
                << " (was " << d_max_cache_size_in_bytes << ").";
        throw InternalErr(__FILE__, __LINE__, msg.str());
    }
#endif
    DBG(cerr << "DAP Cache: directory " << d_cache_dir << ", prefix " << d_prefix
            << ", max size " << d_max_cache_size_in_bytes << endl );
}

void DAPCache3::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 InternalErr 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 InternalErr(__FILE__, __LINE__, "Could not write size info to the cache info file in startup!");

        // 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 InternalErr(__FILE__, __LINE__, get_errno());
        }
    }

    DBG(cerr << "d_cache_info_fd: " << d_cache_info_fd << endl);
}

#if 0

BESCache3::BESCache3(BESKeys *keys, const string &cache_dir_key, const string &prefix_key, const string &size_key) :
        d_max_cache_size_in_bytes(0)
{
    bool found = false;
    keys->get_value(cache_dir_key, d_cache_dir, found);
    if (!found)
        throw BESSyntaxUserError("The cache directory key " + cache_dir_key + " was not found in the BES configuration file", __FILE__, __LINE__);

    found = false;
    keys->get_value(prefix_key, d_prefix, found);
    if (!found)
        throw BESSyntaxUserError("The prefix key " + prefix_key + " was not found in the BES configuration file", __FILE__, __LINE__);

    found = false;
    string cache_size_str;
    keys->get_value(size_key, cache_size_str, found);
    if (!found)
        throw BESSyntaxUserError("The size key " + size_key + " was not found in the BES configuration file", __FILE__, __LINE__);

    std::istringstream is(cache_size_str);
    is >> d_max_cache_size_in_bytes;

    m_initialize_cache_info();
}
#endif


string DAPCache3::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, DAPCache3::DAP_CACHE_CHAR);
        }
        string::size_type last_dot = target.rfind('.');
        if (last_dot != string::npos) {
            target = target.substr(0, last_dot);
        }
    }
    DBG(cerr << "  d_cache_dir: '" << d_cache_dir << "'" << endl);
    DBG(cerr << "  d_prefix:    '" << d_prefix << "'" << endl);
    DBG(cerr << "  target:      '" << target  << "'" << endl);

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

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

    bool status = getSharedLock(target, fd);

    DBG(cerr << "DAP Cache: read_lock: " << target << "(" << status << ")" << endl);

    if (status)
        m_record_descriptor(target, fd);

    unlock_cache();

    return status;
}

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

    bool status = createLockedFile(target, fd);

    DBG(cerr << "DAP Cache: create_and_lock: " << target << "(" << status << ")" << endl);

    if (status)
        m_record_descriptor(target, fd);

    unlock_cache();

    return status;

}

void DAPCache3::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 InternalErr(__FILE__, __LINE__, get_errno());
    }
}

void DAPCache3::lock_cache_write()
{
    DBG(cerr << "lock_cache - d_cache_info_fd: " << d_cache_info_fd << endl);

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

void DAPCache3::lock_cache_read()
{
    DBG(cerr << "lock_cache - d_cache_info_fd: " << d_cache_info_fd << endl);

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

void DAPCache3::unlock_cache()
{
    DBG(cerr << "DAP Cache: unlock: cache_info (fd: " << d_cache_info_fd << ")" << endl);

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

void DAPCache3::unlock_and_close(const string &file_name)
{
    DBG(cerr << "DAP Cache: unlock file: " << file_name << endl);

    unlock(m_get_descriptor(file_name));
}

void DAPCache3::unlock_and_close(int fd)
{
    DBG(cerr << "DAP Cache: unlock fd: " << fd << endl);

    unlock(fd);

    DBG(cerr << "DAP Cache: unlock " << fd << " Success" << endl);
}

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

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

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

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

        DBG(cerr << "DAP Cache: cache size updated to: " << current_size << endl);

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

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

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

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

unsigned long long DAPCache3::get_cache_size()
{
    try {
        lock_cache_read();

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

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


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

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

    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 InternalErr(__FILE__, __LINE__, "Zero-byte file found in cache. " + *file);
#endif
            contents.push_back(entry);
        }
    }

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

    return current_size;
}

void DAPCache3::update_and_purge(const string &new_file)
{
    DBG(cerr << "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" )) {
            DBG(endl << "BEFORE Purge " << computed_size/BYTES_PER_MEG << endl );
            CacheFiles::iterator ti = contents.begin();
            CacheFiles::iterator te = contents.end();
            for (; ti != te; ti++) {
                DBG((*ti).time << ": " << (*ti).name << ": size " << (*ti).size/BYTES_PER_MEG << endl );
            }
        }
#endif
        DBG(cerr << "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)) {
                    DBG(cerr << "purge: " << i->name << " removed." << endl );

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

                    unlock(cfile_fd);
                    computed_size -= i->size;
                }
#if 0
                else {
                    // This information is useful when debugging... Might comment out for production
                    DBG(cerr << "purge: " << i->name << " is in use." << endl );
                }
#endif
                ++i;

                DBG(cerr << "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 InternalErr(__FILE__, __LINE__, "Could not rewind to front of cache info file.");

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

void DAPCache3::purge_file(const string &file)
{
    DBG(cerr << "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;
            }

            DBG(cerr << "purge_file: " << file << " removed." << endl );

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

            unlock(cfile_fd);

            unsigned long long cache_size = get_cache_size() - size;

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

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

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

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

} // namespace libdap