path: root/cpp/src/qpid/legacystore/jrnl/fcntl.cpp

/*
 *
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 fcntl.cpp
 *
 * Qpid asynchronous store plugin library
 *
 * File containing code for class mrg::journal::fcntl (non-logging file
 * handle), used for controlling journal log files. See comments in file
 * fcntl.h for details.
 */

#include "qpid/legacystore/jrnl/fcntl.h"

#include <cerrno>
#include <cstdlib>
#include <cstring>
#include <fcntl.h>
#include <iomanip>
#include "qpid/legacystore/jrnl/jerrno.h"
#include "qpid/legacystore/jrnl/jexception.h"
#include <sstream>
#include <unistd.h>

namespace mrg
{
namespace journal
{

fcntl::fcntl(const std::string& fbasename, const u_int16_t pfid, const u_int16_t lfid, const u_int32_t jfsize_sblks,
        const rcvdat* const ro):
        _fname(),
        _pfid(pfid),
        _lfid(lfid),
        _ffull_dblks(JRNL_SBLK_SIZE * (jfsize_sblks + 1)),
        _wr_fh(-1),
        _rec_enqcnt(0),
        _rd_subm_cnt_dblks(0),
        _rd_cmpl_cnt_dblks(0),
        _wr_subm_cnt_dblks(0),
        _wr_cmpl_cnt_dblks(0),
        _aio_cnt(0),
        _fhdr_wr_aio_outstanding(false)
{
    initialize(fbasename, pfid, lfid, jfsize_sblks, ro);
    open_wr_fh();
}

fcntl::~fcntl()
{
    close_wr_fh();
}

bool
fcntl::reset(const rcvdat* const ro)
{
    rd_reset();
    return wr_reset(ro);
}

void
fcntl::rd_reset()
{
    _rd_subm_cnt_dblks = 0;
    _rd_cmpl_cnt_dblks = 0;
}

bool
fcntl::wr_reset(const rcvdat* const ro)
{
    if (ro)
    {
        if (!ro->_jempty)
        {
            if (ro->_lfid == _pfid)
            {
                _wr_subm_cnt_dblks = ro->_eo/JRNL_DBLK_SIZE;
                _wr_cmpl_cnt_dblks = ro->_eo/JRNL_DBLK_SIZE;
            }
            else
            {
                _wr_subm_cnt_dblks = _ffull_dblks;
                _wr_cmpl_cnt_dblks = _ffull_dblks;
            }
            _rec_enqcnt = ro->_enq_cnt_list[_pfid];
            return true;
        }
    }
    // Journal overflow test - checks if the file to be reset still contains enqueued records
    // or outstanding aios
    if (_rec_enqcnt || _aio_cnt)
        return false;
    _wr_subm_cnt_dblks = 0;
    _wr_cmpl_cnt_dblks = 0;
    return true;
}

int
fcntl::open_wr_fh()
{
    if (_wr_fh < 0)
    {
        _wr_fh = ::open(_fname.c_str(), O_WRONLY | O_DIRECT, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH); // 0644 -rw-r--r--
        if (_wr_fh < 0)
        {
            std::ostringstream oss;
            oss << "pfid=" << _pfid << " lfid=" << _lfid << " file=\"" << _fname << "\"" << FORMAT_SYSERR(errno);
            throw jexception(jerrno::JERR_FCNTL_OPENWR, oss.str(), "fcntl", "open_fh");
        }
    }
    return _wr_fh;
}

void
fcntl::close_wr_fh()
{
    if (_wr_fh >= 0)
    {
        ::close(_wr_fh);
        _wr_fh = -1;
    }
}

u_int32_t
fcntl::add_enqcnt(u_int32_t a)
{
    _rec_enqcnt += a;
    return _rec_enqcnt;
}

u_int32_t
fcntl::decr_enqcnt()
{
    if (_rec_enqcnt == 0)
    {
        std::ostringstream oss;
        oss << "pfid=" << _pfid << " lfid=" << _lfid;
        throw jexception(jerrno::JERR__UNDERFLOW, oss.str(), "fcntl", "decr_enqcnt");
    }
    return --_rec_enqcnt;
}

u_int32_t
fcntl::subtr_enqcnt(u_int32_t s)
{
    if (_rec_enqcnt < s)
    {
        std::ostringstream oss;
        oss << "pfid=" << _pfid << " lfid=" << _lfid << " rec_enqcnt=" << _rec_enqcnt << " decr=" << s;
        throw jexception(jerrno::JERR__UNDERFLOW, oss.str(), "fcntl", "subtr_enqcnt");
    }
    _rec_enqcnt -= s;
    return _rec_enqcnt;
}

u_int32_t
fcntl::add_rd_subm_cnt_dblks(u_int32_t a)
{
    if (_rd_subm_cnt_dblks + a > _wr_subm_cnt_dblks)
    {
        std::ostringstream oss;
        oss << "pfid=" << _pfid << " lfid=" << _lfid << " rd_subm_cnt_dblks=" << _rd_subm_cnt_dblks << " incr=" << a;
        oss << " wr_subm_cnt_dblks=" << _wr_subm_cnt_dblks;
        throw jexception(jerrno::JERR_FCNTL_RDOFFSOVFL, oss.str(), "fcntl", "add_rd_subm_cnt_dblks");
    }
    _rd_subm_cnt_dblks += a;
    return _rd_subm_cnt_dblks;
}

u_int32_t
fcntl::add_rd_cmpl_cnt_dblks(u_int32_t a)
{
    if (_rd_cmpl_cnt_dblks + a > _rd_subm_cnt_dblks)
    {
        std::ostringstream oss;
        oss << "pfid=" << _pfid << " lfid=" << _lfid << " rd_cmpl_cnt_dblks=" << _rd_cmpl_cnt_dblks << " incr=" << a;
        oss << " rd_subm_cnt_dblks=" << _rd_subm_cnt_dblks;
        throw jexception(jerrno::JERR_FCNTL_CMPLOFFSOVFL, oss.str(), "fcntl", "add_rd_cmpl_cnt_dblks");
    }
    _rd_cmpl_cnt_dblks += a;
    return _rd_cmpl_cnt_dblks;
}

u_int32_t
fcntl::add_wr_subm_cnt_dblks(u_int32_t a)
{
    if (_wr_subm_cnt_dblks + a > _ffull_dblks) // Allow for file header
    {
        std::ostringstream oss;
        oss << "pfid=" << _pfid << " lfid=" << _lfid << " wr_subm_cnt_dblks=" << _wr_subm_cnt_dblks << " incr=" << a;
        oss << " fsize=" << _ffull_dblks << " dblks";
        throw jexception(jerrno::JERR_FCNTL_FILEOFFSOVFL, oss.str(), "fcntl", "add_wr_subm_cnt_dblks");
    }
    _wr_subm_cnt_dblks += a;
    return _wr_subm_cnt_dblks;
}

u_int32_t
fcntl::add_wr_cmpl_cnt_dblks(u_int32_t a)
{
    if (_wr_cmpl_cnt_dblks + a > _wr_subm_cnt_dblks)
    {
        std::ostringstream oss;
        oss << "pfid=" << _pfid << " lfid=" << _lfid << " wr_cmpl_cnt_dblks=" << _wr_cmpl_cnt_dblks << " incr=" << a;
        oss << " wr_subm_cnt_dblks=" << _wr_subm_cnt_dblks;
        throw jexception(jerrno::JERR_FCNTL_CMPLOFFSOVFL, oss.str(), "fcntl", "add_wr_cmpl_cnt_dblks");
    }
    _wr_cmpl_cnt_dblks += a;
    return _wr_cmpl_cnt_dblks;
}

u_int16_t
fcntl::decr_aio_cnt()
{
    if(_aio_cnt == 0)
    {
        std::ostringstream oss;
        oss << "pfid=" << _pfid << " lfid=" << _lfid << " Decremented aio_cnt to below zero";
        throw jexception(jerrno::JERR__UNDERFLOW, oss.str(), "fcntl", "decr_aio_cnt");
    }
    return --_aio_cnt;
}

// Debug function
const std::string
fcntl::status_str() const
{
    std::ostringstream oss;
    oss << "pfid=" << _pfid << " ws=" << _wr_subm_cnt_dblks << " wc=" << _wr_cmpl_cnt_dblks;
    oss << " rs=" << _rd_subm_cnt_dblks << " rc=" << _rd_cmpl_cnt_dblks;
    oss << " ec=" << _rec_enqcnt << " ac=" << _aio_cnt;
    return oss.str();
}

// Protected functions

void
fcntl::initialize(const std::string& fbasename, const u_int16_t pfid, const u_int16_t lfid, const u_int32_t jfsize_sblks,
        const rcvdat* const ro)
{
    _pfid = pfid;
    _lfid = lfid;
    _fname = filename(fbasename, pfid);

#ifdef RHM_JOWRITE
    // In test mode, only create file if it does not exist
    struct stat s;
    if (::stat(_fname.c_str(), &s))
    {
#endif
        if (ro) // Recovery initialization: set counters only
        {
            if (!ro->_jempty)
            {
                // For last file only, set write counters to end of last record (the
                // continuation point); for all others, set to eof.
                if (ro->_lfid == _pfid)
                {
                    _wr_subm_cnt_dblks = ro->_eo/JRNL_DBLK_SIZE;
                    _wr_cmpl_cnt_dblks = ro->_eo/JRNL_DBLK_SIZE;
                }
                else
                {
                    _wr_subm_cnt_dblks = _ffull_dblks;
                    _wr_cmpl_cnt_dblks = _ffull_dblks;
                }
                // Set the number of enqueued records for this file.
                _rec_enqcnt = ro->_enq_cnt_list[_pfid];
            }
        }
        else // Normal initialization: create empty journal files
            create_jfile(jfsize_sblks);
#ifdef RHM_JOWRITE
    }
#endif
}

std::string
fcntl::filename(const std::string& fbasename, const u_int16_t pfid)
{
    std::ostringstream oss;
    oss << fbasename << ".";
    oss << std::setw(4) << std::setfill('0') << std::hex << pfid;
    oss << "." << JRNL_DATA_EXTENSION;
    return oss.str();
}

void
fcntl::clean_file(const u_int32_t jfsize_sblks)
{
    // NOTE: The journal file size is always one sblock bigger than the specified journal
    // file size, which is the data content size. The extra block is for the journal file
    // header which precedes all data on each file and is exactly one sblock in size.
    u_int32_t nsblks = jfsize_sblks + 1;

    // TODO - look at more efficient alternatives to allocating a null block:
    // 1. mmap() against /dev/zero, but can alignment for O_DIRECT be assured?
    // 2. ftruncate(), but does this result in a sparse file? If so, then this is no good.

    // Create temp null block for writing
    const std::size_t sblksize = JRNL_DBLK_SIZE * JRNL_SBLK_SIZE;
    void* nullbuf = 0;
    // Allocate no more than 2MB (4096 sblks) as a null buffer
    const u_int32_t nullbuffsize_sblks = nsblks > 4096 ? 4096 : nsblks;
    const std::size_t nullbuffsize = nullbuffsize_sblks * sblksize;
    if (::posix_memalign(&nullbuf, sblksize, nullbuffsize))
    {
        std::ostringstream oss;
        oss << "posix_memalign() failed: size=" << nullbuffsize << " blk_size=" << sblksize;
        oss << FORMAT_SYSERR(errno);
        throw jexception(jerrno::JERR__MALLOC, oss.str(), "fcntl", "clean_file");
    }
    std::memset(nullbuf, 0, nullbuffsize);

    int fh = ::open(_fname.c_str(), O_WRONLY | O_CREAT | O_DIRECT,
            S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH); // 0644 -rw-r--r--
    if (fh < 0)
    {
        std::free(nullbuf);
        std::ostringstream oss;
        oss << "open() failed:" << FORMAT_SYSERR(errno);
        throw jexception(jerrno::JERR_FCNTL_OPENWR, oss.str(), "fcntl", "clean_file");
    }

    while (nsblks > 0)
    {
        u_int32_t this_write_sblks = nsblks >= nullbuffsize_sblks ? nullbuffsize_sblks : nsblks;
        if (::write(fh, nullbuf, this_write_sblks * sblksize) == -1)
        {
            ::close(fh);
            std::free(nullbuf);
            std::ostringstream oss;
            oss << "wr_size=" << (this_write_sblks * sblksize) << FORMAT_SYSERR(errno);
            throw jexception(jerrno::JERR_FCNTL_WRITE, oss.str(), "fcntl", "clean_file");
        }
        nsblks -= this_write_sblks;
    }

    // Clean up
    std::free(nullbuf);
    if (::close(fh))
    {
        std::ostringstream oss;
        oss << FORMAT_SYSERR(errno);
        throw jexception(jerrno::JERR_FCNTL_CLOSE, oss.str(), "fcntl", "clean_file");
    }
}

void
fcntl::create_jfile(const u_int32_t jfsize_sblks)
{
    clean_file(jfsize_sblks);
}

} // namespace journal
} // namespace mrg