Author: kpvdr Date: 2008-07-31 16:58:27 -0400 (Thu, 31 Jul 2008) New Revision: 2242 Added: store/trunk/cpp/lib/jrnl/fcntl.cpp store/trunk/cpp/lib/jrnl/fcntl.hpp Removed: store/trunk/cpp/lib/jrnl/lfh.cpp store/trunk/cpp/lib/jrnl/lfh.hpp store/trunk/cpp/lib/jrnl/nlfh.cpp store/trunk/cpp/lib/jrnl/nlfh.hpp Modified: store/trunk/cpp/lib/Makefile.am store/trunk/cpp/lib/jrnl/jcntl.cpp store/trunk/cpp/lib/jrnl/jcntl.hpp store/trunk/cpp/lib/jrnl/jerrno.cpp store/trunk/cpp/lib/jrnl/jerrno.hpp store/trunk/cpp/lib/jrnl/pmgr.hpp store/trunk/cpp/lib/jrnl/rmgr.cpp store/trunk/cpp/lib/jrnl/rrfc.cpp store/trunk/cpp/lib/jrnl/rrfc.hpp store/trunk/cpp/lib/jrnl/wmgr.cpp store/trunk/cpp/lib/jrnl/wrfc.cpp store/trunk/cpp/lib/jrnl/wrfc.hpp Log: Removed class lfh (logging file handle, derived from class nlfh), an artifact of early design and subequently left unimplemented, but is used in journal. Renamed parent class from nlfh (non-logging file handle) to fcntl (file controller) in preparation to move file handles out of these classes. Other minor tidy-up. Modified: store/trunk/cpp/lib/Makefile.am =================================================================== --- store/trunk/cpp/lib/Makefile.am 2008-07-31 19:02:43 UTC (rev 2241) +++ store/trunk/cpp/lib/Makefile.am 2008-07-31 20:58:27 UTC (rev 2242) @@ -3,107 +3,105 @@ lib_LTLIBRARIES = libbdbstore.la -libbdbstore_la_LIBADD = \ - $(APR_LIBS) \ - $(LIB_DLOPEN) \ - $(LIB_BERKELEY_DB) \ - $(LIB_CLOCK_GETTIME) \ +libbdbstore_la_LIBADD = \ + $(APR_LIBS) \ + $(LIB_DLOPEN) \ + $(LIB_BERKELEY_DB) \ + $(LIB_CLOCK_GETTIME) \ $(QPID_LIBS) -libbdbstore_la_LDFLAGS = \ - -version-info \ +libbdbstore_la_LDFLAGS = \ + -version-info \ $(LIBTOOL_VERSION_INFO_ARG) -libbdbstore_la_SOURCES = \ - StorePlugin.cpp \ - BdbMessageStore.cpp \ - BindingDbt.cpp \ - BufferValue.cpp \ - DataTokenImpl.cpp \ - IdDbt.cpp \ - IdPairDbt.cpp \ - IdSequence.cpp \ - JournalImpl.cpp \ - PreparedTransaction.cpp \ - StringDbt.cpp \ - BdbMessageStore.h \ - BindingDbt.h \ - BufferValue.h \ - Cursor.h \ - DataTokenImpl.h \ - IdDbt.h \ - IdPairDbt.h \ - IdSequence.h \ - JournalImpl.h \ - PreparedTransaction.h \ - StoreException.h \ - StringDbt.h \ - TxnCtxt.h \ +libbdbstore_la_SOURCES = \ + StorePlugin.cpp \ + BdbMessageStore.cpp \ + BindingDbt.cpp \ + BufferValue.cpp \ + DataTokenImpl.cpp \ + IdDbt.cpp \ + IdPairDbt.cpp \ + IdSequence.cpp \ + JournalImpl.cpp \ + PreparedTransaction.cpp \ + StringDbt.cpp \ + BdbMessageStore.h \ + BindingDbt.h \ + BufferValue.h \ + Cursor.h \ + DataTokenImpl.h \ + IdDbt.h \ + IdPairDbt.h \ + IdSequence.h \ + JournalImpl.h \ + PreparedTransaction.h \ + StoreException.h \ + StringDbt.h \ + TxnCtxt.h \ jrnl/aio.cpp \ - jrnl/arr_cnt.cpp \ - jrnl/cvar.cpp \ - jrnl/data_tok.cpp \ - jrnl/deq_rec.cpp \ - jrnl/enq_map.cpp \ - jrnl/enq_rec.cpp \ - jrnl/jcntl.cpp \ - jrnl/jdir.cpp \ - jrnl/jerrno.cpp \ - jrnl/jexception.cpp \ - jrnl/jinf.cpp \ - jrnl/jrec.cpp \ - jrnl/lfh.cpp \ - jrnl/nlfh.cpp \ - jrnl/pmgr.cpp \ - jrnl/rmgr.cpp \ - jrnl/rrfc.cpp \ - jrnl/slock.cpp \ - jrnl/time_ns.cpp \ - jrnl/txn_map.cpp \ - jrnl/txn_rec.cpp \ - jrnl/wmgr.cpp \ - jrnl/wrfc.cpp \ + jrnl/arr_cnt.cpp \ + jrnl/cvar.cpp \ + jrnl/data_tok.cpp \ + jrnl/deq_rec.cpp \ + jrnl/enq_map.cpp \ + jrnl/enq_rec.cpp \ + jrnl/fcntl.cpp \ + jrnl/jcntl.cpp \ + jrnl/jdir.cpp \ + jrnl/jerrno.cpp \ + jrnl/jexception.cpp \ + jrnl/jinf.cpp \ + jrnl/jrec.cpp \ + jrnl/pmgr.cpp \ + jrnl/rmgr.cpp \ + jrnl/rrfc.cpp \ + jrnl/slock.cpp \ + jrnl/time_ns.cpp \ + jrnl/txn_map.cpp \ + jrnl/txn_rec.cpp \ + jrnl/wmgr.cpp \ + jrnl/wrfc.cpp \ jrnl/aio.hpp \ - jrnl/aio_cb.hpp \ - jrnl/arr_cnt.hpp \ - jrnl/cvar.hpp \ - jrnl/data_tok.hpp \ - jrnl/deq_hdr.hpp \ - jrnl/deq_rec.hpp \ - jrnl/enq_hdr.hpp \ - jrnl/enq_map.hpp \ - jrnl/enq_rec.hpp \ - jrnl/enums.hpp \ - jrnl/file_hdr.hpp \ - jrnl/jcfg.hpp \ - jrnl/jcntl.hpp \ - jrnl/jdir.hpp \ - jrnl/jerrno.hpp \ - jrnl/jexception.hpp \ - jrnl/jinf.hpp \ - jrnl/jrec.hpp \ - jrnl/lfh.hpp \ - jrnl/nlfh.hpp \ - jrnl/pmgr.hpp \ - jrnl/rcvdat.hpp \ - jrnl/rec_hdr.hpp \ - jrnl/rec_tail.hpp \ - jrnl/rmgr.hpp \ - jrnl/rrfc.hpp \ - jrnl/slock.hpp \ - jrnl/time_ns.hpp \ - jrnl/txn_hdr.hpp \ - jrnl/txn_map.hpp \ - jrnl/txn_rec.hpp \ - jrnl/wmgr.hpp \ - jrnl/wrfc.hpp \ - gen/qpid/management/PackageMrgstore.cpp \ - gen/qpid/management/PackageMrgstore.h \ - gen/qpid/management/Journal.cpp \ - gen/qpid/management/Journal.h \ - gen/qpid/management/Store.cpp \ - gen/qpid/management/Store.h \ - gen/qpid/management/ArgsJournalExpand.h \ + jrnl/aio_cb.hpp \ + jrnl/arr_cnt.hpp \ + jrnl/cvar.hpp \ + jrnl/data_tok.hpp \ + jrnl/deq_hdr.hpp \ + jrnl/deq_rec.hpp \ + jrnl/enq_hdr.hpp \ + jrnl/enq_map.hpp \ + jrnl/enq_rec.hpp \ + jrnl/enums.hpp \ + jrnl/fcntl.hpp \ + jrnl/file_hdr.hpp \ + jrnl/jcfg.hpp \ + jrnl/jcntl.hpp \ + jrnl/jdir.hpp \ + jrnl/jerrno.hpp \ + jrnl/jexception.hpp \ + jrnl/jinf.hpp \ + jrnl/jrec.hpp \ + jrnl/pmgr.hpp \ + jrnl/rcvdat.hpp \ + jrnl/rec_hdr.hpp \ + jrnl/rec_tail.hpp \ + jrnl/rmgr.hpp \ + jrnl/rrfc.hpp \ + jrnl/slock.hpp \ + jrnl/time_ns.hpp \ + jrnl/txn_hdr.hpp \ + jrnl/txn_map.hpp \ + jrnl/txn_rec.hpp \ + jrnl/wmgr.hpp \ + jrnl/wrfc.hpp \ + gen/qpid/management/PackageMrgstore.cpp \ + gen/qpid/management/PackageMrgstore.h \ + gen/qpid/management/Journal.cpp \ + gen/qpid/management/Journal.h \ + gen/qpid/management/Store.cpp \ + gen/qpid/management/Store.h \ + gen/qpid/management/ArgsJournalExpand.h \ gen/qpid/management/ArgsJournalReconfigure.h Copied: store/trunk/cpp/lib/jrnl/fcntl.cpp (from rev 2238, store/trunk/cpp/lib/jrnl/nlfh.cpp) =================================================================== --- store/trunk/cpp/lib/jrnl/fcntl.cpp (rev 0) +++ store/trunk/cpp/lib/jrnl/fcntl.cpp 2008-07-31 20:58:27 UTC (rev 2242) @@ -0,0 +1,409 @@ +/** +* \file fcntl.cpp +* +* Red Hat Messaging - Message Journal +* +* File containing code for class rhm::journal::fcntl (non-logging file +* handle), used for controlling journal log files. See comments in file +* fcntl.hpp for details. +* +* Copyright 2007, 2008 Red Hat, Inc. +* +* This file is part of Red Hat Messaging. +* +* Red Hat Messaging 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 +* +* The GNU Lesser General Public License is available in the file COPYING. +*/ + +#include "jrnl/fcntl.hpp" + +#include <cerrno> +#include <cstdlib> +#include <cstring> +#include <fcntl.h> +#include <iomanip> +#include "jrnl/jerrno.hpp" +#include "jrnl/jexception.hpp" +#include <sstream> + +namespace rhm +{ +namespace journal +{ + +fcntl::fcntl(const u_int32_t jfsize_sblks, const u_int16_t fid): + _fname(), + _fid(fid), + _ffull_dblks(JRNL_SBLK_SIZE * (jfsize_sblks + 1)), + _rd_fh(-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) +{} + +fcntl::fcntl(const std::string& fbasename, const u_int16_t fid, const u_int32_t jfsize_sblks, + const rcvdat* const ro): + _fname(), + _fid(fid), + _ffull_dblks(JRNL_SBLK_SIZE * (jfsize_sblks + 1)), + _rd_fh(-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) +{ + initialize(fbasename, fid, jfsize_sblks, ro); + open_fh(); +} + +fcntl::~fcntl() +{ + close_fh(); +} + +void +fcntl::initialize(const std::string& fbasename, const u_int16_t fid, + const u_int32_t jfsize_sblks, const rcvdat* const ro) +{ + _fid = fid; + _fname = filename(fbasename, fid); + +#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->_empty) + { + // For last file only, set write counters to end of last record (the + // continuation point); for all others, set to eof. + if (ro->_lfid == _fid) + { + _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[_fid]; + } + } + else // Normal initialization: create empty journal files + create_jfile(jfsize_sblks); +#ifdef RHM_JOWRITE + } +#endif +} + +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->_empty) + { + if (ro->_lfid == _fid) + { + _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[_fid]; + 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; +} + +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) + throw jexception(jerrno::JERR__UNDERFLOW, "fcntl", "decr_enqcnt"); + return --_rec_enqcnt; +} + +u_int32_t +fcntl::subtr_enqcnt(u_int32_t s) +{ + if (_rec_enqcnt < s) + { + std::ostringstream oss; + oss << "_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 << "_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 << "_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 << "_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 << "_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) + throw jexception(jerrno::JERR__UNDERFLOW, "Decremented _aio_cnt when already zero", "fcntl", + "decr_aio_cnt"); + return --_aio_cnt; +} + +// Debug function +const std::string +fcntl::status_str() const +{ + std::ostringstream oss; + oss << "fid=" << _fid << " 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(); +} + +// Private functions + +void +fcntl::open_fh() +{ + _rd_fh = ::open(_fname.c_str(), O_RDONLY | O_DIRECT); + if (_rd_fh < 0) + { + std::ostringstream oss; + oss << "file=\"" << _fname << "\"" << FORMAT_SYSERR(errno); + throw jexception(jerrno::JERR_FCNTL_OPENRD, oss.str(), "fcntl", "open_fh"); + } + _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 << "file=\"" << _fname << "\"" << FORMAT_SYSERR(errno); + throw jexception(jerrno::JERR_FCNTL_OPENWR, oss.str(), "fcntl", "open_fh"); + } +} + +void +fcntl::close_fh() +{ + if (_rd_fh >= 0) + { + ::close(_rd_fh); + _rd_fh = -1; + } + if (_wr_fh >= 0) + { + ::close(_wr_fh); + _wr_fh = -1; + } +} + +std::string +fcntl::filename(const std::string& fbasename, const u_int16_t fid) +{ + std::ostringstream oss; + oss << fbasename << "."; + oss << std::setw(4) << std::setfill('0') << std::hex << fid; + 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); +} + +void +fcntl::get_jfile() +{ + // Step 1: stat _pool_dir, locate first available file, check file size matches + // Step 2: move file to current directory +} + +void return_jfile() +{ + // Step 1: Clean file (if req'd) + // Step 2: Move file to _pool_dir +} + + +// Static declarations + +std::string fcntl::_pool_dir; + +} // namespace journal +} // namespace rhm Copied: store/trunk/cpp/lib/jrnl/fcntl.hpp (from rev 2238, store/trunk/cpp/lib/jrnl/nlfh.hpp) =================================================================== --- store/trunk/cpp/lib/jrnl/fcntl.hpp (rev 0) +++ store/trunk/cpp/lib/jrnl/fcntl.hpp 2008-07-31 20:58:27 UTC (rev 2242) @@ -0,0 +1,160 @@ +/** +* \file fcntl.hpp +* +* Red Hat Messaging - Message Journal +* +* File containing code for class rhm::journal::fcntl (non-logging file +* handle), used for controlling journal log files. See class documentation for +* details. +* +* Copyright 2007, 2008 Red Hat, Inc. +* +* This file is part of Red Hat Messaging. +* +* Red Hat Messaging 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 +* +* The GNU Lesser General Public License is available in the file COPYING. +*/ + +#ifndef rhm_journal_fcntl_hpp +#define rhm_journal_fcntl_hpp + +namespace rhm +{ +namespace journal +{ +class fcntl; +} +} + +#include <cstddef> +#include <string> +#include "jrnl/rcvdat.hpp" +#include <sys/types.h> + +namespace rhm +{ +namespace journal +{ + + /** + * \class fcntl + * \brief Journal file controller. There is one instance per journal file. + */ + class fcntl + { + protected: + static std::string _pool_dir; ///< Directory containing unused file pool + std::string _fname; ///< File name + u_int16_t _fid; ///< File ID (ordinal number in ring buffer) + const u_int32_t _ffull_dblks; ///< File size in dblks (incl. file header) + int _rd_fh; ///< Read file handle + int _wr_fh; ///< Write file handle + u_int32_t _rec_enqcnt; ///< Count of enqueued records + u_int32_t _rd_subm_cnt_dblks; ///< Read file count (data blocks) for submitted AIO + u_int32_t _rd_cmpl_cnt_dblks; ///< Read file count (data blocks) for completed AIO + u_int32_t _wr_subm_cnt_dblks; ///< Write file count (data blocks) for submitted AIO + u_int32_t _wr_cmpl_cnt_dblks; ///< Write file count (data blocks) for completed AIO + u_int16_t _aio_cnt; ///< Outstanding AIO operations on this file + + public: + fcntl(const u_int32_t jfsize_sblks, const u_int16_t fid); + // Constructors with implicit initialize() and open() + fcntl(const std::string& fbasename, const u_int16_t fid, const u_int32_t jfsize_sblks, + const rcvdat* const ro); + virtual ~fcntl(); + + inline static std::string& pool_dir() { return _pool_dir; } + inline static void set_pool_dir(const std::string pool_dir) { _pool_dir = pool_dir; } + + virtual void initialize(const std::string& fbasename, const u_int16_t fid, + const u_int32_t jfsize_sblks, const rcvdat* const ro); + virtual bool reset(const rcvdat* const ro = 0); + virtual void rd_reset(); + virtual bool wr_reset(const rcvdat* const ro = 0); + + inline const std::string& fname() const { return _fname; } + inline u_int16_t fid() const { return _fid; } + inline int rd_fh() const { return _rd_fh; } + inline int wr_fh() const { return _wr_fh; } + inline u_int32_t enqcnt() const { return _rec_enqcnt; } + inline u_int32_t incr_enqcnt() { return ++_rec_enqcnt; } + u_int32_t add_enqcnt(u_int32_t a); + u_int32_t decr_enqcnt(); + u_int32_t subtr_enqcnt(u_int32_t s); + + inline u_int32_t rd_subm_cnt_dblks() const { return _rd_subm_cnt_dblks; } + inline std::size_t rd_subm_offs() const { return _rd_subm_cnt_dblks * JRNL_DBLK_SIZE; } + u_int32_t add_rd_subm_cnt_dblks(u_int32_t a); + + inline u_int32_t rd_cmpl_cnt_dblks() const { return _rd_cmpl_cnt_dblks; } + inline std::size_t rd_cmpl_offs() const { return _rd_cmpl_cnt_dblks * JRNL_DBLK_SIZE; } + u_int32_t add_rd_cmpl_cnt_dblks(u_int32_t a); + + inline u_int32_t wr_subm_cnt_dblks() const { return _wr_subm_cnt_dblks; } + inline std::size_t wr_subm_offs() const { return _wr_subm_cnt_dblks * JRNL_DBLK_SIZE; } + u_int32_t add_wr_subm_cnt_dblks(u_int32_t a); + + inline u_int32_t wr_cmpl_cnt_dblks() const { return _wr_cmpl_cnt_dblks; } + inline std::size_t wr_cmpl_offs() const { return _wr_cmpl_cnt_dblks * JRNL_DBLK_SIZE; } + u_int32_t add_wr_cmpl_cnt_dblks(u_int32_t a); + + inline u_int16_t aio_cnt() const { return _aio_cnt; } + inline u_int16_t incr_aio_cnt() { return ++_aio_cnt; } + u_int16_t decr_aio_cnt(); + + // Derived helper functions + + inline bool rd_void() const { return _wr_cmpl_cnt_dblks == 0; } + inline bool rd_empty() const { return _wr_cmpl_cnt_dblks <= JRNL_SBLK_SIZE; } + inline u_int32_t rd_remaining_dblks() const + { return _wr_cmpl_cnt_dblks - _rd_subm_cnt_dblks; } + inline bool is_rd_full() const { return _wr_cmpl_cnt_dblks == _rd_subm_cnt_dblks; } + inline bool is_rd_compl() const + { return _wr_cmpl_cnt_dblks == _rd_cmpl_cnt_dblks; } + inline u_int32_t rd_aio_outstanding_dblks() const + { return _rd_subm_cnt_dblks - _rd_cmpl_cnt_dblks; } + inline bool rd_file_rotate() const { return is_rd_full() && is_wr_compl(); } + + inline bool wr_void() const { return _wr_subm_cnt_dblks == 0; } + inline bool wr_empty() const { return _wr_subm_cnt_dblks <= JRNL_SBLK_SIZE; } + inline u_int32_t wr_remaining_dblks() const + { return _ffull_dblks - _wr_subm_cnt_dblks; } + inline bool is_wr_full() const { return _ffull_dblks == _wr_subm_cnt_dblks; } + inline bool is_wr_compl() const { return _ffull_dblks == _wr_cmpl_cnt_dblks; } + inline u_int32_t wr_aio_outstanding_dblks() const + { return _wr_subm_cnt_dblks - _wr_cmpl_cnt_dblks; } + inline bool wr_file_rotate() const { return is_wr_full(); } + + // Debug aid + const std::string status_str() const; + + protected: + virtual void open_fh(); + virtual void close_fh(); + static std::string filename(const std::string& fbasename, const u_int16_t fid); + void clean_file(const u_int32_t jfsize_sblks); + void create_jfile(const u_int32_t jfsize_sblks); + + // These functions for future use of file pool + void get_jfile(); + void return_jfile(); + }; + +} // namespace journal +} // namespace rhm + +#endif // ifndef rhm_journal_fcntl_hpp Modified: store/trunk/cpp/lib/jrnl/jcntl.cpp =================================================================== --- store/trunk/cpp/lib/jrnl/jcntl.cpp 2008-07-31 19:02:43 UTC (rev 2241) +++ store/trunk/cpp/lib/jrnl/jcntl.cpp 2008-07-31 20:58:27 UTC (rev 2242) @@ -63,7 +63,7 @@ _autostop(true), _num_jfiles(0), _jfsize_sblks(0), - _datafh(0), + _fc_arr(0), _emap(), _tmap(), _rrfc(), @@ -81,11 +81,11 @@ if (_init_flag && !_stop_flag) try { stop(true); } catch (const jexception& e) { std::cerr << e << std::endl; } - if (_datafh) + if (_fc_arr) { for (u_int32_t i=0; i<_num_jfiles; i++) - delete _datafh[i]; - delete[] _datafh; + delete _fc_arr[i]; + delete[] _fc_arr; } ::pthread_mutex_destroy(&_gev_mutex); ::pthread_mutex_destroy(&_wr_mutex); @@ -104,11 +104,11 @@ _tmap.clear(); // TODO - place this in a finalize() fn? - see ~jcntl() & recover()... - if (_datafh) + if (_fc_arr) { for (u_int32_t i=0; i<_num_jfiles; i++) - delete _datafh[i]; - delete[] _datafh; + delete _fc_arr[i]; + delete[] _fc_arr; } // Set new file geometry parameters @@ -125,22 +125,19 @@ // Clear any existing journal files _jdir.clear_dir(); - _datafh = new lfh*[_num_jfiles]; + _fc_arr = new fcntl*[_num_jfiles]; // Zero the pointer array first because new() can throw exceptions - std::memset(_datafh, 0, sizeof(lfh*) * _num_jfiles); for (u_int16_t i=0; i<_num_jfiles; i++) + _fc_arr[i] = 0; + for (u_int16_t i=0; i<_num_jfiles; i++) { std::ostringstream oss; oss << _jdir << "/" << _base_filename; - _datafh[i] = new lfh(oss.str(), i, _jfsize_sblks, 0); + _fc_arr[i] = new fcntl(oss.str(), i, _jfsize_sblks, 0); } - // TODO: Check the following comment/note (may be obsolete): - // NOTE: The write RFC must initialize first. This sets all the file handle object - // (lfh) counters and pointers for both read and write, since write activity - // constrains read activity (i.e. one can't read what has not yet been written). - _wrfc.initialize(_num_jfiles, _jfsize_sblks, (nlfh**)_datafh); - _rrfc.initialize(_num_jfiles, (nlfh**)_datafh); + _wrfc.initialize(_num_jfiles, _jfsize_sblks, _fc_arr); + _rrfc.initialize(_num_jfiles, _fc_arr); _rmgr.initialize(rd_cb); _wmgr.initialize(wr_cb, wcache_pgsize_sblks, wcache_num_pages, JRNL_WMGR_MAXDTOKPP, JRNL_WMGR_MAXWAITUS); @@ -165,11 +162,11 @@ _tmap.clear(); // TODO - place this in a finalize() fn? - see ~jcntl() & initialize()... - if (_datafh) + if (_fc_arr) { for (u_int32_t i=0; i<_num_jfiles; i++) - delete _datafh[i]; - delete[] _datafh; + delete _fc_arr[i]; + delete[] _fc_arr; } // Set new file geometry parameters @@ -194,22 +191,19 @@ this->log(LOG_DEBUG, _rcvdat.to_log(_jid)); - _datafh = new lfh*[_num_jfiles]; + _fc_arr = new fcntl*[_num_jfiles]; // 0 the pointer array first because new() can throw exceptions - std::memset(_datafh, 0, sizeof(lfh*) * _num_jfiles); for (u_int16_t i=0; i<_num_jfiles; i++) + _fc_arr[i] = 0; + for (u_int16_t i=0; i<_num_jfiles; i++) { std::ostringstream oss; oss << _jdir << "/" << _base_filename; - _datafh[i] = new lfh(oss.str(), i, _jfsize_sblks, &_rcvdat); + _fc_arr[i] = new fcntl(oss.str(), i, _jfsize_sblks, &_rcvdat); } - // TODO: Check the following comment/note (may be obsolete): - // NOTE: The write RFC must initialize first. This sets all the file handle object - // (lfh) counters and pointers for both read and write, since write activity - // constrains read activity (i.e. one can't read what has not yet been written). - _wrfc.initialize(_num_jfiles, _jfsize_sblks, (nlfh**)_datafh, &_rcvdat); - _rrfc.initialize(_num_jfiles, (nlfh**)_datafh, _rcvdat._ffid); + _wrfc.initialize(_num_jfiles, _jfsize_sblks, _fc_arr, &_rcvdat); + _rrfc.initialize(_num_jfiles, _fc_arr, _rcvdat._ffid); _rmgr.initialize(rd_cb); _wmgr.initialize(wr_cb, wcache_pgsize_sblks, wcache_num_pages, JRNL_WMGR_MAXDTOKPP, JRNL_WMGR_MAXWAITUS, _rcvdat._eo); @@ -224,9 +218,9 @@ if (!_readonly_flag) throw jexception(jerrno::JERR_JCNTL_NOTRECOVERED, "jcntl", "recover_complete"); for (u_int16_t i=0; i<_num_jfiles; i++) - _datafh[i]->reset(&_rcvdat); - _wrfc.initialize(_num_jfiles, _jfsize_sblks, (nlfh**)_datafh, &_rcvdat); - _rrfc.initialize(_num_jfiles, (nlfh**)_datafh, _rcvdat._ffid); + _fc_arr[i]->reset(&_rcvdat); + _wrfc.initialize(_num_jfiles, _jfsize_sblks, _fc_arr, &_rcvdat); + _rrfc.initialize(_num_jfiles, _fc_arr, _rcvdat._ffid); _rmgr.recover_complete(); _readonly_flag = false; } Modified: store/trunk/cpp/lib/jrnl/jcntl.hpp =================================================================== --- store/trunk/cpp/lib/jrnl/jcntl.hpp 2008-07-31 19:02:43 UTC (rev 2241) +++ store/trunk/cpp/lib/jrnl/jcntl.hpp 2008-07-31 20:58:27 UTC (rev 2242) @@ -42,7 +42,7 @@ #include <cstddef> #include <deque> #include "jrnl/jdir.hpp" -#include "jrnl/lfh.hpp" +#include "jrnl/fcntl.hpp" #include "jrnl/rcvdat.hpp" #include "jrnl/rmgr.hpp" #include "jrnl/wmgr.hpp" @@ -130,7 +130,7 @@ // Journal control structures u_int16_t _num_jfiles; ///< Number of journal files u_int32_t _jfsize_sblks; ///< Journal file size in sblks - lfh** _datafh; ///< Array of pointers to data file handles + fcntl** _fc_arr; ///< Array of pointers to data file controllers enq_map _emap; ///< Enqueue map for low water mark management txn_map _tmap; ///< Transaction map open transactions rrfc _rrfc; ///< Read journal rotating file controller @@ -540,13 +540,13 @@ { return _wrfc.aio_outstanding_dblks(); } inline u_int32_t get_wr_outstanding_aio_dblks(u_int16_t pi) const - { return _wrfc.file_handle(pi)->wr_aio_outstanding_dblks(); } + { return _wrfc.file_controller(pi)->wr_aio_outstanding_dblks(); } inline u_int32_t get_rd_outstanding_aio_dblks() const { return _rrfc.aio_outstanding_dblks(); } inline u_int32_t get_rd_outstanding_aio_dblks(u_int16_t pi) const - { return _rrfc.file_handle(pi)->rd_aio_outstanding_dblks(); } + { return _rrfc.file_controller(pi)->rd_aio_outstanding_dblks(); } inline u_int16_t get_rd_fid() const { return _rrfc.index(); } inline u_int16_t get_wr_fid() const { return _wrfc.index(); } Modified: store/trunk/cpp/lib/jrnl/jerrno.cpp =================================================================== --- store/trunk/cpp/lib/jrnl/jerrno.cpp 2008-07-31 19:02:43 UTC (rev 2241) +++ store/trunk/cpp/lib/jrnl/jerrno.cpp 2008-07-31 20:58:27 UTC (rev 2242) @@ -74,14 +74,14 @@ const u_int32_t jerrno::JERR_JDIR_UNLINK = 0x0309; const u_int32_t jerrno::JERR_JDIR_BADFTYPE = 0x030a; -// class nlfh -const u_int32_t jerrno::JERR_NLFH_OPENRD = 0x0400; -const u_int32_t jerrno::JERR_NLFH_OPENWR = 0x0401; -const u_int32_t jerrno::JERR_NLFH_WRITE = 0x0402; -const u_int32_t jerrno::JERR_NLFH_CLOSE = 0x0403; -const u_int32_t jerrno::JERR_NLFH_FILEOFFSOVFL = 0x0404; -const u_int32_t jerrno::JERR_NFLH_CMPLOFFSOVFL = 0x0405; -const u_int32_t jerrno::JERR_NFLH_RDOFFSOVFL = 0x0406; +// class fcntl +const u_int32_t jerrno::JERR_FCNTL_OPENRD = 0x0400; +const u_int32_t jerrno::JERR_FCNTL_OPENWR = 0x0401; +const u_int32_t jerrno::JERR_FCNTL_WRITE = 0x0402; +const u_int32_t jerrno::JERR_FCNTL_CLOSE = 0x0403; +const u_int32_t jerrno::JERR_FCNTL_FILEOFFSOVFL = 0x0404; +const u_int32_t jerrno::JERR_FCNTL_CMPLOFFSOVFL = 0x0405; +const u_int32_t jerrno::JERR_FCNTL_RDOFFSOVFL = 0x0406; // class file_hdr @@ -159,16 +159,16 @@ _err_map[JERR_JDIR_UNLINK] = "JERR_JDIR_UNLINK: File delete failed."; _err_map[JERR_JDIR_BADFTYPE] = "JERR_JDIR_BADFTYPE: Bad or unknown file type (stat mode)."; - // class nlfh - _err_map[JERR_NLFH_OPENRD] = "JERR_JNLFH_OPENRD: Unable to open file for read."; - _err_map[JERR_NLFH_OPENWR] = "JERR_JNLFH_OPENWR: Unable to open file for write."; - _err_map[JERR_NLFH_WRITE] = "JERR_JNLFH_WRITE: Unable to write to file."; - _err_map[JERR_NLFH_CLOSE] = "JERR_JNLFH_CLOSE: File close failed."; - _err_map[JERR_NLFH_FILEOFFSOVFL] = "JERR_NLFH_FILEOFFSOVFL: " + // class fcntl + _err_map[JERR_FCNTL_OPENRD] = "JERR_FCNTL_OPENRD: Unable to open file for read."; + _err_map[JERR_FCNTL_OPENWR] = "JERR_FCNTL_OPENWR: Unable to open file for write."; + _err_map[JERR_FCNTL_WRITE] = "JERR_FCNTL_WRITE: Unable to write to file."; + _err_map[JERR_FCNTL_CLOSE] = "JERR_FCNTL_CLOSE: File close failed."; + _err_map[JERR_FCNTL_FILEOFFSOVFL] = "JERR_FCNTL_FILEOFFSOVFL: " "Attempted increase file offset past file size."; - _err_map[JERR_NFLH_CMPLOFFSOVFL] = "JERR_NFLH_CMPLOFFSOVFL: " + _err_map[JERR_FCNTL_CMPLOFFSOVFL] = "JERR_FCNTL_CMPLOFFSOVFL: " "Attempted increase completed file offset past submitted offset."; - _err_map[JERR_NFLH_RDOFFSOVFL] = "JERR_NFLH_RDOFFSOVFL: " + _err_map[JERR_FCNTL_RDOFFSOVFL] = "JERR_FCNTL_RDOFFSOVFL: " "Attempted increase read offset past write offset."; Modified: store/trunk/cpp/lib/jrnl/jerrno.hpp =================================================================== --- store/trunk/cpp/lib/jrnl/jerrno.hpp 2008-07-31 19:02:43 UTC (rev 2241) +++ store/trunk/cpp/lib/jrnl/jerrno.hpp 2008-07-31 20:58:27 UTC (rev 2242) @@ -92,14 +92,14 @@ static const u_int32_t JERR_JDIR_UNLINK; ///< File delete failed static const u_int32_t JERR_JDIR_BADFTYPE; ///< Bad or unknown file type (stat mode) - // class nlfh - static const u_int32_t JERR_NLFH_OPENRD; ///< Unable to open file for read - static const u_int32_t JERR_NLFH_OPENWR; ///< Unable to open file for write - static const u_int32_t JERR_NLFH_WRITE; ///< Unable to write to file - static const u_int32_t JERR_NLFH_CLOSE; ///< File close failed - static const u_int32_t JERR_NLFH_FILEOFFSOVFL; ///< Increased offset past file size - static const u_int32_t JERR_NFLH_CMPLOFFSOVFL; ///< Increased cmpl offs past subm offs - static const u_int32_t JERR_NFLH_RDOFFSOVFL; ///< Increased read offs past write offs + // class fcntl + static const u_int32_t JERR_FCNTL_OPENRD; ///< Unable to open file for read + static const u_int32_t JERR_FCNTL_OPENWR; ///< Unable to open file for write + static const u_int32_t JERR_FCNTL_WRITE; ///< Unable to write to file + static const u_int32_t JERR_FCNTL_CLOSE; ///< File close failed + static const u_int32_t JERR_FCNTL_FILEOFFSOVFL; ///< Increased offset past file size + static const u_int32_t JERR_FCNTL_CMPLOFFSOVFL; ///< Increased cmpl offs past subm offs + static const u_int32_t JERR_FCNTL_RDOFFSOVFL; ///< Increased read offs past write offs // class file_hdr Deleted: store/trunk/cpp/lib/jrnl/lfh.cpp =================================================================== --- store/trunk/cpp/lib/jrnl/lfh.cpp 2008-07-31 19:02:43 UTC (rev 2241) +++ store/trunk/cpp/lib/jrnl/lfh.cpp 2008-07-31 20:58:27 UTC (rev 2242) @@ -1,55 +0,0 @@ -/** -* \file lfh.cpp -* -* Red Hat Messaging - Message Journal -* -* File containing code for class rhm::journal::lfh (logging file -* handle), used for controlling journal data files. See comments in file -* lfh.hpp for details. -* -* \author Kim van der Riet -* -* Copyright 2007 Red Hat, Inc. -* -* This file is part of Red Hat Messaging. -* -* Red Hat Messaging 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 -* -* The GNU Lesser General Public License is available in the file COPYING. -*/ - -#include "jrnl/lfh.hpp" - -namespace rhm -{ -namespace journal -{ - -lfh::lfh(const u_int32_t jfsize_sblks, const u_int16_t fid): - nlfh(jfsize_sblks, fid) -{} - -lfh::lfh(const std::string& fbasename, const u_int16_t fid, const u_int32_t jfsize_sblks, - rcvdat const * const ro): - nlfh(fbasename, fid, jfsize_sblks, ro) -{} - -lfh::~lfh() -{} - -} // namespace journal -} // namespace rhm - Deleted: store/trunk/cpp/lib/jrnl/lfh.hpp =================================================================== --- store/trunk/cpp/lib/jrnl/lfh.hpp 2008-07-31 19:02:43 UTC (rev 2241) +++ store/trunk/cpp/lib/jrnl/lfh.hpp 2008-07-31 20:58:27 UTC (rev 2242) @@ -1,68 +0,0 @@ -/** -* \file lfh.hpp -* -* Red Hat Messaging - Message Journal -* -* File containing code for class rhm::journal::lfh (logging file -* handle), used for controlling journal data files. See class -* documentation for details. -* -* \author Kim van der Riet -* -* Copyright 2007 Red Hat, Inc. -* -* This file is part of Red Hat Messaging. -* -* Red Hat Messaging 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 -* -* The GNU Lesser General Public License is available in the file COPYING. -*/ - -#ifndef rhm_journal_lfh_hpp -#define rhm_journal_lfh_hpp - -namespace rhm -{ -namespace journal -{ -class lfh; -} -} - -#include "jrnl/nlfh.hpp" - -namespace rhm -{ -namespace journal -{ - /** - * \class lfh - * \brief Logging journal file handler. This means that each data action is logged. - */ - class lfh : public nlfh - { - public: - lfh(const u_int32_t jfsize_sblks, const u_int16_t fid); - lfh(const std::string& fbasename, const u_int16_t fid, const u_int32_t jfsize_sblks, - rcvdat const * const ro); - virtual ~lfh(); - }; - -} // namespace journal -} // namespace rhm - -#endif // ifndef rhm_journal_lfh_hpp - Deleted: store/trunk/cpp/lib/jrnl/nlfh.cpp =================================================================== --- store/trunk/cpp/lib/jrnl/nlfh.cpp 2008-07-31 19:02:43 UTC (rev 2241) +++ store/trunk/cpp/lib/jrnl/nlfh.cpp 2008-07-31 20:58:27 UTC (rev 2242) @@ -1,409 +0,0 @@ -/** -* \file nlfh.cpp -* -* Red Hat Messaging - Message Journal -* -* File containing code for class rhm::journal::nlfh (non-logging file -* handle), used for controlling journal log files. See comments in file -* nlfh.hpp for details. -* -* Copyright 2007, 2008 Red Hat, Inc. -* -* This file is part of Red Hat Messaging. -* -* Red Hat Messaging 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 -* -* The GNU Lesser General Public License is available in the file COPYING. -*/ - -#include "jrnl/nlfh.hpp" - -#include <cerrno> -#include <cstdlib> -#include <cstring> -#include <fcntl.h> -#include <iomanip> -#include "jrnl/jerrno.hpp" -#include "jrnl/jexception.hpp" -#include <sstream> - -namespace rhm -{ -namespace journal -{ - -nlfh::nlfh(const u_int32_t jfsize_sblks, const u_int16_t fid): - _fname(), - _fid(fid), - _ffull_dblks(JRNL_SBLK_SIZE * (jfsize_sblks + 1)), - _rd_fh(-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) -{} - -nlfh::nlfh(const std::string& fbasename, const u_int16_t fid, const u_int32_t jfsize_sblks, - const rcvdat* const ro): - _fname(), - _fid(fid), - _ffull_dblks(JRNL_SBLK_SIZE * (jfsize_sblks + 1)), - _rd_fh(-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) -{ - initialize(fbasename, fid, jfsize_sblks, ro); - open_fh(); -} - -nlfh::~nlfh() -{ - close_fh(); -} - -void -nlfh::initialize(const std::string& fbasename, const u_int16_t fid, - const u_int32_t jfsize_sblks, const rcvdat* const ro) -{ - _fid = fid; - _fname = filename(fbasename, fid); - -#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->_empty) - { - // For last file only, set write counters to end of last record (the - // continuation point); for all others, set to eof. - if (ro->_lfid == _fid) - { - _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[_fid]; - } - } - else // Normal initialization: create empty journal files - create_jfile(jfsize_sblks); -#ifdef RHM_JOWRITE - } -#endif -} - -bool -nlfh::reset(const rcvdat* const ro) -{ - rd_reset(); - return wr_reset(ro); -} - -void -nlfh::rd_reset() -{ - _rd_subm_cnt_dblks = 0; - _rd_cmpl_cnt_dblks = 0; -} - -bool -nlfh::wr_reset(const rcvdat* const ro) -{ - if (ro) - { - if (!ro->_empty) - { - if (ro->_lfid == _fid) - { - _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[_fid]; - 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; -} - -u_int32_t -nlfh::add_enqcnt(u_int32_t a) -{ - _rec_enqcnt += a; - return _rec_enqcnt; -} - -u_int32_t -nlfh::decr_enqcnt() -{ - if (_rec_enqcnt == 0) - throw jexception(jerrno::JERR__UNDERFLOW, "nlfh", "decr_enqcnt"); - return --_rec_enqcnt; -} - -u_int32_t -nlfh::subtr_enqcnt(u_int32_t s) -{ - if (_rec_enqcnt < s) - { - std::ostringstream oss; - oss << "_rec_enqcnt=" << _rec_enqcnt << " decr=" << s; - throw jexception(jerrno::JERR__UNDERFLOW, oss.str(), "nlfh", "subtr_enqcnt"); - } - _rec_enqcnt -= s; - return _rec_enqcnt; -} - -u_int32_t -nlfh::add_rd_subm_cnt_dblks(u_int32_t a) -{ - if (_rd_subm_cnt_dblks + a > _wr_subm_cnt_dblks) - { - std::ostringstream oss; - oss << "_rd_subm_cnt_dblks=" << _rd_subm_cnt_dblks << " incr=" << a; - oss << " _wr_subm_cnt_dblks=" << _wr_subm_cnt_dblks; - throw jexception(jerrno::JERR_NFLH_RDOFFSOVFL, oss.str(), "nlfh", "add_rd_subm_cnt_dblks"); - } - _rd_subm_cnt_dblks += a; - return _rd_subm_cnt_dblks; -} - -u_int32_t -nlfh::add_rd_cmpl_cnt_dblks(u_int32_t a) -{ - if (_rd_cmpl_cnt_dblks + a > _rd_subm_cnt_dblks) - { - std::ostringstream oss; - oss << "_rd_cmpl_cnt_dblks=" << _rd_cmpl_cnt_dblks << " incr=" << a; - oss << " _rd_subm_cnt_dblks=" << _rd_subm_cnt_dblks; - throw jexception(jerrno::JERR_NFLH_CMPLOFFSOVFL, oss.str(), "nlfh", - "add_rd_cmpl_cnt_dblks"); - } - _rd_cmpl_cnt_dblks += a; - return _rd_cmpl_cnt_dblks; -} - -u_int32_t -nlfh::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 << "_wr_subm_cnt_dblks=" << _wr_subm_cnt_dblks << " incr=" << a; - oss << " fsize=" << _ffull_dblks << " dblks"; - throw jexception(jerrno::JERR_NLFH_FILEOFFSOVFL, oss.str(), "nlfh", - "add_wr_subm_cnt_dblks"); - } - _wr_subm_cnt_dblks += a; - return _wr_subm_cnt_dblks; -} - -u_int32_t -nlfh::add_wr_cmpl_cnt_dblks(u_int32_t a) -{ - if (_wr_cmpl_cnt_dblks + a > _wr_subm_cnt_dblks) - { - std::ostringstream oss; - oss << "_wr_cmpl_cnt_dblks=" << _wr_cmpl_cnt_dblks << " incr=" << a; - oss << " _wr_subm_cnt_dblks=" << _wr_subm_cnt_dblks; - throw jexception(jerrno::JERR_NFLH_CMPLOFFSOVFL, oss.str(), "nlfh", - "add_wr_cmpl_cnt_dblks"); - } - _wr_cmpl_cnt_dblks += a; - return _wr_cmpl_cnt_dblks; -} - -u_int16_t -nlfh::decr_aio_cnt() -{ - if(_aio_cnt == 0) - throw jexception(jerrno::JERR__UNDERFLOW, "Decremented _aio_cnt when already zero", "nlfh", - "decr_aio_cnt"); - return --_aio_cnt; -} - -// Debug function -const std::string -nlfh::status_str() const -{ - std::ostringstream oss; - oss << "fid=" << _fid << " 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(); -} - -// Private functions - -void -nlfh::open_fh() -{ - _rd_fh = ::open(_fname.c_str(), O_RDONLY | O_DIRECT); - if (_rd_fh < 0) - { - std::ostringstream oss; - oss << "file=\"" << _fname << "\"" << FORMAT_SYSERR(errno); - throw jexception(jerrno::JERR_NLFH_OPENRD, oss.str(), "nlfh", "open_fh"); - } - _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 << "file=\"" << _fname << "\"" << FORMAT_SYSERR(errno); - throw jexception(jerrno::JERR_NLFH_OPENWR, oss.str(), "nlfh", "open_fh"); - } -} - -void -nlfh::close_fh() -{ - if (_rd_fh >= 0) - { - ::close(_rd_fh); - _rd_fh = -1; - } - if (_wr_fh >= 0) - { - ::close(_wr_fh); - _wr_fh = -1; - } -} - -std::string -nlfh::filename(const std::string& fbasename, const u_int16_t fid) -{ - std::ostringstream oss; - oss << fbasename << "."; - oss << std::setw(4) << std::setfill('0') << std::hex << fid; - oss << "." << JRNL_DATA_EXTENSION; - return oss.str(); -} - -void -nlfh::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(), "nlfh", "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_NLFH_OPENWR, oss.str(), "nlfh", "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_NLFH_WRITE, oss.str(), "nlfh", "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_NLFH_CLOSE, oss.str(), "nlfh", "clean_file"); - } -} - -void -nlfh::create_jfile(const u_int32_t jfsize_sblks) -{ - clean_file(jfsize_sblks); -} - -void -nlfh::get_jfile() -{ - // Step 1: stat _pool_dir, locate first available file, check file size matches - // Step 2: move file to current directory -} - -void return_jfile() -{ - // Step 1: Clean file (if req'd) - // Step 2: Move file to _pool_dir -} - - -// Static declarations - -std::string nlfh::_pool_dir; - -} // namespace journal -} // namespace rhm Deleted: store/trunk/cpp/lib/jrnl/nlfh.hpp =================================================================== --- store/trunk/cpp/lib/jrnl/nlfh.hpp 2008-07-31 19:02:43 UTC (rev 2241) +++ store/trunk/cpp/lib/jrnl/nlfh.hpp 2008-07-31 20:58:27 UTC (rev 2242) @@ -1,160 +0,0 @@ -/** -* \file nlfh.hpp -* -* Red Hat Messaging - Message Journal -* -* File containing code for class rhm::journal::nlfh (non-logging file -* handle), used for controlling journal log files. See class documentation for -* details. -* -* Copyright 2007, 2008 Red Hat, Inc. -* -* This file is part of Red Hat Messaging. -* -* Red Hat Messaging 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 -* -* The GNU Lesser General Public License is available in the file COPYING. -*/ - -#ifndef rhm_journal_nlfh_hpp -#define rhm_journal_nlfh_hpp - -namespace rhm -{ -namespace journal -{ -class nlfh; -} -} - -#include <cstddef> -#include <string> -#include "jrnl/rcvdat.hpp" -#include <sys/types.h> - -namespace rhm -{ -namespace journal -{ - - /** - * \class nlfh - * \brief Non-logging file handler. This means that each data action is not logged. - */ - class nlfh - { - protected: - static std::string _pool_dir; ///< Directory containing unused file pool - std::string _fname; ///< File name - u_int16_t _fid; ///< File ID (ordinal number in ring buffer) - const u_int32_t _ffull_dblks; ///< File size in dblks (incl. file header) - int _rd_fh; ///< Read file handle - int _wr_fh; ///< Write file handle - u_int32_t _rec_enqcnt; ///< Count of enqueued records - u_int32_t _rd_subm_cnt_dblks; ///< Read file count (data blocks) for submitted AIO - u_int32_t _rd_cmpl_cnt_dblks; ///< Read file count (data blocks) for completed AIO - u_int32_t _wr_subm_cnt_dblks; ///< Write file count (data blocks) for submitted AIO - u_int32_t _wr_cmpl_cnt_dblks; ///< Write file count (data blocks) for completed AIO - u_int16_t _aio_cnt; ///< Outstanding AIO operations on this file - - public: - nlfh(const u_int32_t jfsize_sblks, const u_int16_t fid); - // Constructors with implicit initialize() and open() - nlfh(const std::string& fbasename, const u_int16_t fid, const u_int32_t jfsize_sblks, - const rcvdat* const ro); - virtual ~nlfh(); - - inline static std::string& pool_dir() { return _pool_dir; } - inline static void set_pool_dir(const std::string pool_dir) { _pool_dir = pool_dir; } - - virtual void initialize(const std::string& fbasename, const u_int16_t fid, - const u_int32_t jfsize_sblks, const rcvdat* const ro); - virtual bool reset(const rcvdat* const ro = 0); - virtual void rd_reset(); - virtual bool wr_reset(const rcvdat* const ro = 0); - - inline const std::string& fname() const { return _fname; } - inline u_int16_t fid() const { return _fid; } - inline int rd_fh() const { return _rd_fh; } - inline int wr_fh() const { return _wr_fh; } - inline u_int32_t enqcnt() const { return _rec_enqcnt; } - inline u_int32_t incr_enqcnt() { return ++_rec_enqcnt; } - u_int32_t add_enqcnt(u_int32_t a); - u_int32_t decr_enqcnt(); - u_int32_t subtr_enqcnt(u_int32_t s); - - inline u_int32_t rd_subm_cnt_dblks() const { return _rd_subm_cnt_dblks; } - inline std::size_t rd_subm_offs() const { return _rd_subm_cnt_dblks * JRNL_DBLK_SIZE; } - u_int32_t add_rd_subm_cnt_dblks(u_int32_t a); - - inline u_int32_t rd_cmpl_cnt_dblks() const { return _rd_cmpl_cnt_dblks; } - inline std::size_t rd_cmpl_offs() const { return _rd_cmpl_cnt_dblks * JRNL_DBLK_SIZE; } - u_int32_t add_rd_cmpl_cnt_dblks(u_int32_t a); - - inline u_int32_t wr_subm_cnt_dblks() const { return _wr_subm_cnt_dblks; } - inline std::size_t wr_subm_offs() const { return _wr_subm_cnt_dblks * JRNL_DBLK_SIZE; } - u_int32_t add_wr_subm_cnt_dblks(u_int32_t a); - - inline u_int32_t wr_cmpl_cnt_dblks() const { return _wr_cmpl_cnt_dblks; } - inline std::size_t wr_cmpl_offs() const { return _wr_cmpl_cnt_dblks * JRNL_DBLK_SIZE; } - u_int32_t add_wr_cmpl_cnt_dblks(u_int32_t a); - - inline u_int16_t aio_cnt() const { return _aio_cnt; } - inline u_int16_t incr_aio_cnt() { return ++_aio_cnt; } - u_int16_t decr_aio_cnt(); - - // Derived helper functions - - inline bool rd_void() const { return _wr_cmpl_cnt_dblks == 0; } - inline bool rd_empty() const { return _wr_cmpl_cnt_dblks <= JRNL_SBLK_SIZE; } - inline u_int32_t rd_remaining_dblks() const - { return _wr_cmpl_cnt_dblks - _rd_subm_cnt_dblks; } - inline bool is_rd_full() const { return _wr_cmpl_cnt_dblks == _rd_subm_cnt_dblks; } - inline bool is_rd_compl() const - { return _wr_cmpl_cnt_dblks == _rd_cmpl_cnt_dblks; } - inline u_int32_t rd_aio_outstanding_dblks() const - { return _rd_subm_cnt_dblks - _rd_cmpl_cnt_dblks; } - inline bool rd_file_rotate() const { return is_rd_full() && is_wr_compl(); } - - inline bool wr_void() const { return _wr_subm_cnt_dblks == 0; } - inline bool wr_empty() const { return _wr_subm_cnt_dblks <= JRNL_SBLK_SIZE; } - inline u_int32_t wr_remaining_dblks() const - { return _ffull_dblks - _wr_subm_cnt_dblks; } - inline bool is_wr_full() const { return _ffull_dblks == _wr_subm_cnt_dblks; } - inline bool is_wr_compl() const { return _ffull_dblks == _wr_cmpl_cnt_dblks; } - inline u_int32_t wr_aio_outstanding_dblks() const - { return _wr_subm_cnt_dblks - _wr_cmpl_cnt_dblks; } - inline bool wr_file_rotate() const { return is_wr_full(); } - - // Debug aid - const std::string status_str() const; - - protected: - virtual void open_fh(); - virtual void close_fh(); - static std::string filename(const std::string& fbasename, const u_int16_t fid); - void clean_file(const u_int32_t jfsize_sblks); - void create_jfile(const u_int32_t jfsize_sblks); - - // These functions for future use of file pool - void get_jfile(); - void return_jfile(); - }; - -} // namespace journal -} // namespace rhm - -#endif // ifndef rhm_journal_nlfh_hpp Modified: store/trunk/cpp/lib/jrnl/pmgr.hpp =================================================================== --- store/trunk/cpp/lib/jrnl/pmgr.hpp 2008-07-31 19:02:43 UTC (rev 2241) +++ store/trunk/cpp/lib/jrnl/pmgr.hpp 2008-07-31 20:58:27 UTC (rev 2242) @@ -46,7 +46,7 @@ #include "jrnl/deq_rec.hpp" #include "jrnl/enq_map.hpp" #include "jrnl/enq_rec.hpp" -#include "jrnl/nlfh.hpp" +#include "jrnl/fcntl.hpp" #include "jrnl/txn_map.hpp" #include "jrnl/txn_rec.hpp" @@ -86,8 +86,8 @@ u_int32_t _wdblks; ///< Total number of dblks in page so far u_int32_t _rdblks; ///< Total number of dblks in page so far std::deque<data_tok*>* _pdtokl; ///< Page message tokens list - nlfh* _wfh; ///< File handle for incrementing write compl counts - nlfh* _rfh; ///< File handle for incrementing read compl counts + fcntl* _wfh; ///< File handle for incrementing write compl counts + fcntl* _rfh; ///< File handle for incrementing read compl counts void* _pbuff; ///< Page buffer page_cb(u_int16_t index); ///< Convenience constructor Modified: store/trunk/cpp/lib/jrnl/rmgr.cpp =================================================================== --- store/trunk/cpp/lib/jrnl/rmgr.cpp 2008-07-31 19:02:43 UTC (rev 2241) +++ store/trunk/cpp/lib/jrnl/rmgr.cpp 2008-07-31 20:58:27 UTC (rev 2242) @@ -291,7 +291,7 @@ { // Increment the completed read offset // NOTE: We cannot use _rrfc here, as it may have rotated since submitting count. - // Use stored pointer to nlfh in the pcb instead. + // Use stored pointer to fcntl in the pcb instead. pcbp->_rdblks = aiocbp->u.c.nbytes / JRNL_DBLK_SIZE; pcbp->_rfh->add_rd_cmpl_cnt_dblks(pcbp->_rdblks); pcbp->_state = state; @@ -596,7 +596,7 @@ _rrfc.add_subm_cnt_dblks(rd_size); _aio_evt_rem++; _page_cb_arr[pi]._state = AIO_PENDING; - _page_cb_arr[pi]._rfh = _rrfc.file_handle(); + _page_cb_arr[pi]._rfh = _rrfc.file_controller(); } else // If there is nothing to read for this page, neither will there be for the others... break; @@ -652,7 +652,7 @@ void rmgr::init_file_header_read(u_int16_t fid) { - int rfh = _rrfc.file_handle(fid)->rd_fh(); + int rfh = _rrfc.file_controller(fid)->rd_fh(); aio::prep_pread_2(_fhdr_aio_cb_ptr, rfh, _fhdr_buffer, _sblksize, 0); if (aio::submit(_ioctx, 1, &_fhdr_aio_cb_ptr) < 0) throw jexception(jerrno::JERR__AIO, "rmgr", "init_file_header_read"); Modified: store/trunk/cpp/lib/jrnl/rrfc.cpp =================================================================== --- store/trunk/cpp/lib/jrnl/rrfc.cpp 2008-07-31 19:02:43 UTC (rev 2241) +++ store/trunk/cpp/lib/jrnl/rrfc.cpp 2008-07-31 20:58:27 UTC (rev 2242) @@ -42,27 +42,27 @@ rrfc::rrfc(): _nfiles(0), - _fh_arr(0), - _fh_index(0), - _curr_fh(0) + _fc_arr(0), + _fc_index(0), + _curr_fc(0) {} rrfc::~rrfc() {} void -rrfc::initialize(const u_int16_t nfiles, nlfh** fh_arr, u_int32_t fh_index) +rrfc::initialize(const u_int16_t nfiles, fcntl** fc_arr, u_int32_t fc_index) { _nfiles = nfiles; - _fh_arr = fh_arr; - reset(fh_index); + _fc_arr = fc_arr; + reset(fc_index); } void -rrfc::reset(u_int32_t fh_index) +rrfc::reset(u_int32_t fc_index) { - _fh_index = fh_index; - _curr_fh = _fh_arr[_fh_index]; - _curr_fh->rd_reset(); + _fc_index = fc_index; + _curr_fc = _fc_arr[_fc_index]; + _curr_fc->rd_reset(); } bool @@ -70,27 +70,27 @@ { if (!_nfiles) throw jexception(jerrno::JERR__NINIT, "rrfc", "rotate"); - u_int16_t next_fh_index = _fh_index + 1; - if (next_fh_index == _nfiles) - next_fh_index = 0; - nlfh* next_fh = _fh_arr[next_fh_index]; - _fh_index = next_fh_index; - _curr_fh = next_fh; + u_int16_t next_fc_index = _fc_index + 1; + if (next_fc_index == _nfiles) + next_fc_index = 0; + fcntl* next_fc = _fc_arr[next_fc_index]; + _fc_index = next_fc_index; + _curr_fc = next_fc; return true; } -nlfh* -rrfc::file_handle(u_int16_t pg_index) const +fcntl* +rrfc::file_controller(u_int16_t pg_index) const { assert(pg_index < _nfiles); - return _fh_arr[pg_index]; + return _fc_arr[pg_index]; } std::string rrfc::status_str() const { std::ostringstream oss; - oss << "rrfc[" << _fh_index << "]: " << _curr_fh->status_str(); + oss << "rrfc[" << _fc_index << "]: " << _curr_fc->status_str(); return oss.str(); } Modified: store/trunk/cpp/lib/jrnl/rrfc.hpp =================================================================== --- store/trunk/cpp/lib/jrnl/rrfc.hpp 2008-07-31 19:02:43 UTC (rev 2241) +++ store/trunk/cpp/lib/jrnl/rrfc.hpp 2008-07-31 20:58:27 UTC (rev 2242) @@ -42,7 +42,7 @@ } #include <cstddef> -#include "jrnl/nlfh.hpp" +#include "jrnl/fcntl.hpp" namespace rhm { @@ -51,15 +51,15 @@ /** * \class rrfc - * \brief Class to handle manangement of a journal rotating file handle. + * \brief Class to handle read manangement of a journal rotating file controller. */ class rrfc { protected: u_int16_t _nfiles; ///< Number of data files - nlfh** _fh_arr; ///< Array of pointers to data file handles - u_int16_t _fh_index; ///< Index of current file handle - nlfh* _curr_fh; ///< Pointer to current file handle + fcntl** _fc_arr; ///< Array of pointers to data file controllers + u_int16_t _fc_index; ///< Index of current file controller + fcntl* _curr_fc; ///< Pointer to current file controller public: rrfc(); @@ -68,16 +68,16 @@ /** * \brief Initialize the controller. * \param nfiles Number of files in the rotating file group. - * \param fh_arr Pointer to an array of file handles (nlogging_fh or subclasses), - * each of which correspond to one of the physical files. - * \param fh_index Initial index of journal file. Default = 0. + * \param fc_arr Pointer to an array of file controller, each of which correspond to one of + * the physical journal files. + * \param fc_index Initial index of journal file. Default = 0. */ - virtual void initialize(const u_int16_t nfiles, nlfh** fh_arr, u_int32_t fh_index = 0); + virtual void initialize(const u_int16_t nfiles, fcntl** fc_arr, u_int32_t fc_index = 0); - void reset(u_int32_t fh_index = 0); + void reset(u_int32_t fc_index = 0); /** - * \brief Rotate active file handle to next file in rotating file group. + * \brief Rotate active file controller to next file in rotating file group. * \exception jerrno::JERR__NINIT if called before calling initialize(). */ bool rotate(); @@ -86,54 +86,54 @@ * \brief Returns the index of the currently active file within the rotating * file group. */ - inline u_int16_t index() const { return _fh_index; } + inline u_int16_t index() const { return _fc_index; } /** - * \brief Returns the currently active journal file handle within the rotating + * \brief Returns the currently active journal file controller within the rotating * file group. */ - inline nlfh* file_handle() const { return _curr_fh; } + inline fcntl* file_controller() const { return _curr_fc; } /** - * \brief Returns the journal file handle for the given page index within the rotating + * \brief Returns the journal file controller for the given page index within the rotating * file group. */ - nlfh* file_handle(u_int16_t pg_index) const; + fcntl* file_controller(u_int16_t pg_index) const; - // Convenience access methods to current file handle + // Convenience access methods to current file controller - inline u_int16_t fid() const { return _curr_fh->fid(); } - inline int fh() const { return _curr_fh->rd_fh(); } - inline u_int32_t enqcnt() const { return _curr_fh->enqcnt(); } - inline u_int32_t incr_enqcnt() { return _curr_fh->incr_enqcnt(); } - inline u_int32_t incr_enqcnt(u_int16_t fid) { return _fh_arr[fid]->incr_enqcnt(); } - inline u_int32_t add_enqcnt(u_int32_t a) { return _curr_fh->add_enqcnt(a); } + inline u_int16_t fid() const { return _curr_fc->fid(); } + inline int fh() const { return _curr_fc->rd_fh(); } + inline u_int32_t enqcnt() const { return _curr_fc->enqcnt(); } + inline u_int32_t incr_enqcnt() { return _curr_fc->incr_enqcnt(); } + inline u_int32_t incr_enqcnt(u_int16_t fid) { return _fc_arr[fid]->incr_enqcnt(); } + inline u_int32_t add_enqcnt(u_int32_t a) { return _curr_fc->add_enqcnt(a); } inline u_int32_t add_enqcnt(u_int16_t fid, u_int32_t a) - { return _fh_arr[fid]->add_enqcnt(a); } - inline u_int32_t decr_enqcnt(u_int16_t fid) { return _fh_arr[fid]->decr_enqcnt(); } + { return _fc_arr[fid]->add_enqcnt(a); } + inline u_int32_t decr_enqcnt(u_int16_t fid) { return _fc_arr[fid]->decr_enqcnt(); } inline u_int32_t subtr_enqcnt(u_int16_t fid, u_int32_t s) - { return _fh_arr[fid]->subtr_enqcnt(s); } + { return _fc_arr[fid]->subtr_enqcnt(s); } - inline u_int32_t subm_cnt_dblks() const { return _curr_fh->rd_subm_cnt_dblks(); } - inline std::size_t subm_offs() const { return _curr_fh->rd_subm_offs(); } + inline u_int32_t subm_cnt_dblks() const { return _curr_fc->rd_subm_cnt_dblks(); } + inline std::size_t subm_offs() const { return _curr_fc->rd_subm_offs(); } inline u_int32_t add_subm_cnt_dblks(u_int32_t a) - { return _curr_fh->add_rd_subm_cnt_dblks(a); } + { return _curr_fc->add_rd_subm_cnt_dblks(a); } - inline u_int32_t cmpl_cnt_dblks() const { return _curr_fh->rd_cmpl_cnt_dblks(); } - inline std::size_t cmpl_offs() const { return _curr_fh->rd_cmpl_offs(); } + inline u_int32_t cmpl_cnt_dblks() const { return _curr_fc->rd_cmpl_cnt_dblks(); } + inline std::size_t cmpl_offs() const { return _curr_fc->rd_cmpl_offs(); } inline u_int32_t add_cmpl_cnt_dblks(u_int32_t a) - { return _curr_fh->add_rd_cmpl_cnt_dblks(a); } + { return _curr_fc->add_rd_cmpl_cnt_dblks(a); } - inline bool is_void() const { return _curr_fh->rd_void(); } - inline bool is_empty() const { return _curr_fh->rd_empty(); } - inline u_int32_t remaining_dblks() { return _curr_fh->rd_remaining_dblks(); } - inline bool is_full() const { return _curr_fh->is_rd_full(); } - inline bool is_compl() const { return _curr_fh->is_rd_compl(); } + inline bool is_void() const { return _curr_fc->rd_void(); } + inline bool is_empty() const { return _curr_fc->rd_empty(); } + inline u_int32_t remaining_dblks() { return _curr_fc->rd_remaining_dblks(); } + inline bool is_full() const { return _curr_fc->is_rd_full(); } + inline bool is_compl() const { return _curr_fc->is_rd_compl(); } inline u_int32_t aio_outstanding_dblks() const - { return _curr_fh->rd_aio_outstanding_dblks(); } - inline bool file_rotate() const { return _curr_fh->rd_file_rotate(); } + { return _curr_fc->rd_aio_outstanding_dblks(); } + inline bool file_rotate() const { return _curr_fc->rd_file_rotate(); } inline bool is_wr_aio_outstanding() const - { return _curr_fh->wr_aio_outstanding_dblks() > 0; } + { return _curr_fc->wr_aio_outstanding_dblks() > 0; } // Debug aid std::string status_str() const; Modified: store/trunk/cpp/lib/jrnl/wmgr.cpp =================================================================== --- store/trunk/cpp/lib/jrnl/wmgr.cpp 2008-07-31 19:02:43 UTC (rev 2241) +++ store/trunk/cpp/lib/jrnl/wmgr.cpp 2008-07-31 20:58:27 UTC (rev 2242) @@ -713,7 +713,7 @@ _wrfc.subm_offs()); page_cb* pcbp = (page_cb*)(aiocbp->data); // This page control block (pcb) pcbp->_wdblks = _cached_offset_dblks; - pcbp->_wfh = _wrfc.file_handle(); + pcbp->_wfh = _wrfc.file_controller(); if (aio::submit(_ioctx, 1, &aiocbp) < 0) throw jexception(jerrno::JERR__AIO, "wmgr", "write_flush"); _wrfc.add_subm_cnt_dblks(_cached_offset_dblks); @@ -856,7 +856,7 @@ // Increment the completed write offset // NOTE: We cannot use _wrfc here, as it may have rotated since submitting count. - // Use stored pointer to nlfh in the pcb instead. + // Use stored pointer to fcntl in the pcb instead. pcbp->_wfh->add_wr_cmpl_cnt_dblks(pcbp->_wdblks); pcbp->_wfh->decr_aio_cnt(); _jc->instr_decr_outstanding_aio_cnt(); @@ -874,8 +874,8 @@ // get fid from original file header record, update pointers for that fid file_hdr* fhp = (file_hdr*)aiocbp->u.c.buf; u_int32_t fid = fhp->_fid; - nlfh* nlfhp = _wrfc.file_handle(fid); - nlfhp->add_wr_cmpl_cnt_dblks(JRNL_SBLK_SIZE); + fcntl* fcntlp = _wrfc.file_controller(fid); + fcntlp->add_wr_cmpl_cnt_dblks(JRNL_SBLK_SIZE); } } Modified: store/trunk/cpp/lib/jrnl/wrfc.cpp =================================================================== --- store/trunk/cpp/lib/jrnl/wrfc.cpp 2008-07-31 19:02:43 UTC (rev 2241) +++ store/trunk/cpp/lib/jrnl/wrfc.cpp 2008-07-31 20:58:27 UTC (rev 2242) @@ -58,15 +58,15 @@ wrfc::~wrfc() {} void -wrfc::initialize(const u_int16_t nfiles, const u_int32_t fsize_sblks, nlfh** fh_arr, rcvdat* rdp) +wrfc::initialize(const u_int16_t nfiles, const u_int32_t fsize_sblks, fcntl** fc_arr, rcvdat* rdp) { if (rdp) { _nfiles = nfiles; - _fh_arr = fh_arr; - _fh_index = rdp->_lfid; - _curr_fh = _fh_arr[rdp->_lfid]; - _curr_fh->wr_reset(rdp); + _fc_arr = fc_arr; + _fc_index = rdp->_lfid; + _curr_fc = _fc_arr[rdp->_lfid]; + _curr_fc->wr_reset(rdp); _rid = rdp->_h_rid + 1; _reset_ok = true; _owi = rdp->_owi; @@ -76,7 +76,7 @@ } else { - rrfc::initialize(nfiles, fh_arr); + rrfc::initialize(nfiles, fc_arr); #ifdef DRHM_TESTVALS _rid = u_int64_t(0xffeeddcc) << 32; #else @@ -98,15 +98,15 @@ { if (!_nfiles) throw jexception(jerrno::JERR__NINIT, "wrfc", "rotate"); - _fh_index++; - if (_fh_index == _nfiles) + _fc_index++; + if (_fc_index == _nfiles) { - _fh_index = 0; + _fc_index = 0; _owi = !_owi; _frot = false; } - _curr_fh = _fh_arr[_fh_index]; - if (_curr_fh->aio_cnt()) + _curr_fc = _fc_arr[_fc_index]; + if (_curr_fc->aio_cnt()) return RHM_IORES_FILE_AIOWAIT; if (!reset()) //Checks if file is still in use (ie not fully dequeued yet) return RHM_IORES_FULL; @@ -118,7 +118,7 @@ { if (_frot) return 0; - u_int16_t next_index = _fh_index + 1; + u_int16_t next_index = _fc_index + 1; if (next_index >= _nfiles) next_index = 0; return next_index; @@ -127,7 +127,7 @@ bool wrfc::reset() { - _reset_ok = _curr_fh->reset(); // returns false if full (ie file still contains enqueued recs) + _reset_ok = _curr_fc->reset(); // returns false if full (ie file still contains enqueued recs) return _reset_ok; } @@ -158,29 +158,29 @@ if (subm_dblks != 0) subm_dblks -= 4; u_int32_t fwd_dblks = subm_dblks + enq_dsize_dblks + _enq_cap_offs_dblks; - u_int16_t findex = _fh_index; - nlfh* fhp = _curr_fh; + u_int16_t findex = _fc_index; + fcntl* fcp = _curr_fc; bool in_use = false; - while (fwd_dblks && !(findex != _fh_index && fhp->enqcnt())) + while (fwd_dblks && !(findex != _fc_index && fcp->enqcnt())) { fwd_dblks -= fwd_dblks > _fsize_dblks ? _fsize_dblks : fwd_dblks; if (fwd_dblks) { if (++findex == _nfiles) findex = 0; - fhp = _fh_arr[findex]; + fcp = _fc_arr[findex]; } - in_use |= fhp->enqcnt() > 0; + in_use |= fcp->enqcnt() > 0; } // Return true if threshold exceeded - return findex != _fh_index && in_use; + return findex != _fc_index && in_use; } std::string wrfc::status_str() const { std::ostringstream oss; - oss << "wrfc[" << _fh_index << "]: " << _curr_fh->status_str(); + oss << "wrfc[" << _fc_index << "]: " << _curr_fc->status_str(); return oss.str(); } Modified: store/trunk/cpp/lib/jrnl/wrfc.hpp =================================================================== --- store/trunk/cpp/lib/jrnl/wrfc.hpp 2008-07-31 19:02:43 UTC (rev 2241) +++ store/trunk/cpp/lib/jrnl/wrfc.hpp 2008-07-31 20:58:27 UTC (rev 2242) @@ -50,19 +50,18 @@ /** * \class wrfc - * \brief Class to handle manangement of a journal rotating file handle. + * \brief Class to handle write manangement of a journal rotating file controller. */ class wrfc : public rrfc { private: - u_int32_t _fsize_sblks; ///< Size of journal files in sblks - u_int32_t _fsize_dblks; ///< Size of journal files in dblks + u_int32_t _fsize_sblks; ///< Size of journal files in sblks + u_int32_t _fsize_dblks; ///< Size of journal files in dblks u_int32_t _enq_cap_offs_dblks; ///< Enqueue capacity offset - u_int64_t _rid; ///< Master counter for record ID (rid) - bool _reset_ok; ///< Flag set when reset succeeds - bool _owi; ///< Overwrite indicator - bool _frot; ///< Flag is true for first rotation, false at all other times - + u_int64_t _rid; ///< Master counter for record ID (rid) + bool _reset_ok; ///< Flag set when reset succeeds + bool _owi; ///< Overwrite indicator + bool _frot; ///< Flag is true for first rotation, false otherwise public: wrfc(); virtual ~wrfc(); @@ -71,16 +70,16 @@ * \brief Initialize the controller. * \param nfiles Number of files in the rotating file group. * \param fsize_sblks Size of each journal file in sblks. - * \param fh_arr Pointer to an array of file handles (nlogging_fh or subclasses), - * each of which correspond to one of the physical files. + * \param fc_arr Pointer to an array of file controllers, each of which correspond to one of + * the physical journal files. * \param rdp Struct carrying restore information. Optional for non-restore use, defaults to * 0 (NULL). */ - void initialize(const u_int16_t nfiles, const u_int32_t fsize_sblks, nlfh** fh_arr, + void initialize(const u_int16_t nfiles, const u_int32_t fsize_sblks, fcntl** fc_arr, rcvdat* rdp = 0); /** - * \brief Rotate active file handle to next file in rotating file group. + * \brief Rotate active file controller to next file in rotating file group. * \exception jerrno::JERR__NINIT if called before calling initialize(). */ iores rotate(); @@ -99,32 +98,32 @@ inline bool owi() const { return _owi; } inline bool frot() const { return _frot; } - // Convenience access methods to current file handle + // Convenience access methods to current file controller - inline int fh() const { return _curr_fh->wr_fh(); } + inline int fh() const { return _curr_fc->wr_fh(); } - inline u_int32_t subm_cnt_dblks() const { return _curr_fh->wr_subm_cnt_dblks(); } - inline std::size_t subm_offs() const { return _curr_fh->wr_subm_offs(); } + inline u_int32_t subm_cnt_dblks() const { return _curr_fc->wr_subm_cnt_dblks(); } + inline std::size_t subm_offs() const { return _curr_fc->wr_subm_offs(); } inline u_int32_t add_subm_cnt_dblks(u_int32_t a) - { return _curr_fh->add_wr_subm_cnt_dblks(a); } + { return _curr_fc->add_wr_subm_cnt_dblks(a); } - inline u_int32_t cmpl_cnt_dblks() const { return _curr_fh->wr_cmpl_cnt_dblks(); } - inline std::size_t cmpl_offs() const { return _curr_fh->wr_cmpl_offs(); } + inline u_int32_t cmpl_cnt_dblks() const { return _curr_fc->wr_cmpl_cnt_dblks(); } + inline std::size_t cmpl_offs() const { return _curr_fc->wr_cmpl_offs(); } inline u_int32_t add_cmpl_cnt_dblks(u_int32_t a) - { return _curr_fh->add_wr_cmpl_cnt_dblks(a); } + { return _curr_fc->add_wr_cmpl_cnt_dblks(a); } - inline u_int16_t aio_cnt() const { return _curr_fh->aio_cnt(); } - inline u_int16_t incr_aio_cnt() { return _curr_fh->incr_aio_cnt(); } - inline u_int16_t decr_aio_cnt() { return _curr_fh->decr_aio_cnt(); } + inline u_int16_t aio_cnt() const { return _curr_fc->aio_cnt(); } + inline u_int16_t incr_aio_cnt() { return _curr_fc->incr_aio_cnt(); } + inline u_int16_t decr_aio_cnt() { return _curr_fc->decr_aio_cnt(); } - inline bool is_void() const { return _curr_fh->wr_void(); } - inline bool is_empty() const { return _curr_fh->wr_empty(); } - inline u_int32_t remaining_dblks() const { return _curr_fh->wr_remaining_dblks(); } - inline bool is_full() const { return _curr_fh->is_wr_full(); }; - inline bool is_compl() const { return _curr_fh->is_wr_compl(); }; + inline bool is_void() const { return _curr_fc->wr_void(); } + inline bool is_empty() const { return _curr_fc->wr_empty(); } + inline u_int32_t remaining_dblks() const { return _curr_fc->wr_remaining_dblks(); } + inline bool is_full() const { return _curr_fc->is_wr_full(); }; + inline bool is_compl() const { return _curr_fc->is_wr_compl(); }; inline u_int32_t aio_outstanding_dblks() const - { return _curr_fh->wr_aio_outstanding_dblks(); } - inline bool file_rotate() const { return _curr_fh->wr_file_rotate(); } + { return _curr_fc->wr_aio_outstanding_dblks(); } + inline bool file_rotate() const { return _curr_fc->wr_file_rotate(); } bool enq_threshold(const u_int32_t enq_dsize_dblks) const; // Debug aid