July 2008 - rhmessaging-commits - Jboss List Archives

rhmessaging commits: r2242 - in store/trunk/cpp/lib: jrnl and 1 other directory.

by rhmessaging-commits＠lists.jboss.org

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

15 years, 10 months

1
0
0 / 0

rhmessaging commits: r2241 - in store/branches/mrg-1.0/cpp: rhel4-support and 1 other directory.

by rhmessaging-commits＠lists.jboss.org

Author: kpvdr Date: 2008-07-31 15:02:43 -0400 (Thu, 31 Jul 2008) New Revision: 2241 Added: store/branches/mrg-1.0/cpp/rhel4-support/ store/branches/mrg-1.0/cpp/rhel4-support/Makefile store/branches/mrg-1.0/cpp/rhel4-support/README store/branches/mrg-1.0/cpp/rhel4-support/rhel4.patch Log: Added dir for RHEL4 support on 1.0 branch Added: store/branches/mrg-1.0/cpp/rhel4-support/Makefile =================================================================== --- store/branches/mrg-1.0/cpp/rhel4-support/Makefile (rev 0) +++ store/branches/mrg-1.0/cpp/rhel4-support/Makefile 2008-07-31 19:02:43 UTC (rev 2241) @@ -0,0 +1,7 @@ +.PHONY: default apply + +default: + @echo "Run 'make apply' to patch the source for RHEL4" + +apply: rhel4.patch + patch -d .. -p0 < rhel4.patch Added: store/branches/mrg-1.0/cpp/rhel4-support/README =================================================================== --- store/branches/mrg-1.0/cpp/rhel4-support/README (rev 0) +++ store/branches/mrg-1.0/cpp/rhel4-support/README 2008-07-31 19:02:43 UTC (rev 2241) @@ -0,0 +1,8 @@ +To compile on RHEL4, run: + +make apply + +This will apply all necessary patches. Change dirs back to cpp +and run ./bootstrap and ./configure as ususal. Make sure that if +./configure is run with --with-qpid-checkout, it points to a version +of qpid which has also been patched for RHEL4. Added: store/branches/mrg-1.0/cpp/rhel4-support/rhel4.patch =================================================================== --- store/branches/mrg-1.0/cpp/rhel4-support/rhel4.patch (rev 0) +++ store/branches/mrg-1.0/cpp/rhel4-support/rhel4.patch 2008-07-31 19:02:43 UTC (rev 2241) @@ -0,0 +1,276 @@ +Index: tests/.valgrind.supp +=================================================================== +--- tests/.valgrind.supp (revision 2179) ++++ tests/.valgrind.supp (working copy) +@@ -1,32 +1,205 @@ + { +- Benign error in libcpg. +- Memcheck:Param +- socketcall.sendmsg(msg.msg_iov[i]) +- obj:*/libpthread-2.5.so +- obj:*/libcpg.so.2.0.0 ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:_Znwm ++ fun:_ZNSs4_Rep9_S_createEmmRKSaIcE ++ obj:/usr/lib64/libstdc++.so.6.0.3 ++ fun:_ZNSsC1EPKcRKSaIcE ++ fun:_ZN4qpid34options_description_less_easy_initclEPKcPKN5boost15program_options14value_semanticES2_ ++ fun:_ZN4qpid3log7OptionsC1ERKSsS3_ ++ fun:_ZN4qpid3log6LoggerC1Ev + } +- + { +- Uninitialised value problem in _dl_relocate (F7, F8) +- Memcheck:Cond +- fun:_dl_relocate_object +- fun:*dl_* ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:_Znwm ++ fun:_ZNSt6vectorISsSaISsEE13_M_insert_auxEN9__gnu_cxx17__normal_iteratorIPSsS1_EERKSs ++ fun:_ZN4qpid7Options14register_namesESs ++ fun:_ZN4qpid34options_description_less_easy_initclEPKcPKN5boost15program_options14value_semanticES2_ ++ fun:_ZN4qpid3log7OptionsC1ERKSsS3_ ++ fun:_ZN4qpid3log6LoggerC1Ev + } +- + { +- False "possibly leaked" in boost program_options - global std::string var. ++ <insert a suppression name here> + Memcheck:Leak +- fun:_Znwj +- fun:_ZNSs4_Rep9_S_createEjjRKSaIcE +- obj:/usr/lib/libstdc++.so.6.0.8 +- fun:_ZNSsC1EPKcRKSaIcE +- obj:/usr/lib/libboost_program_options.so.1.33.1 ++ fun:_Znwm ++ fun:_ZN5boost15program_options29options_description_easy_initclEPKcPKNS0_14value_semanticES3_ ++ fun:_ZN4qpid34options_description_less_easy_initclEPKcPKN5boost15program_options14value_semanticES2_ ++ fun:_ZN4qpid3log7OptionsC1ERKSsS3_ ++ fun:_ZN4qpid3log6LoggerC1Ev + } ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:calloc ++ fun:_dl_allocate_tls ++ fun:pthread_create@@GLIBC_2.2.5 ++ fun:_ZN3rhm3jtt13jrnl_instance6run_tcEv ++ fun:_Z13constructor_1v ++ fun:_ZN5boost9unit_test9ut_detail17unit_test_monitor8functionEv ++ obj:/usr/lib64/libboost_unit_test_framework.so.1.32.0 ++ fun:_ZN5boost17execution_monitor7executeEbi ++ fun:_ZN5boost9unit_test9ut_detail17unit_test_monitor21execute_and_translateEPNS0_9test_caseEMS3_FvvEi ++ fun:_ZN5boost9unit_test9test_case3runEv ++ fun:_ZN5boost9unit_test10test_suite6do_runEv ++ fun:_ZN5boost9unit_test9test_case3runEv ++ fun:main ++} ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:_Znwm ++ fun:_ZN5boost9unit_test9test_caseC2ENS0_13basic_cstringIKcEEbmb ++} ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:calloc ++ fun:_dl_allocate_tls ++ fun:pthread_create@@GLIBC_2.2.5 ++ fun:_ZN4qpid6broker5Timer5startEv ++ fun:_ZN4qpid6broker5TimerC1Ev ++} + + { +- Bogus epoll_ctl error on i386 +- Memcheck:Param +- epoll_ctl(event) +- fun:epoll_ctl ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:malloc ++ fun:decompose_rpath ++ fun:_dl_map_object ++ fun:openaux ++ fun:_dl_catch_error ++ fun:_dl_map_object_deps ++ fun:dl_open_worker ++ fun:_dl_catch_error ++ fun:_dl_open ++ fun:dlopen_doit ++ fun:_dl_catch_error ++ fun:_dlerror_run ++ fun:dlopen@@GLIBC_2.2.5 ++ fun:_ZN4qpid3sys5Shlib4loadEPKc ++ fun:_Z8tryShlibPKcb ++ fun:main + } ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:calloc ++ fun:_dlerror_run ++ fun:dlopen@@GLIBC_2.2.5 ++ fun:_ZN4qpid3sys5Shlib4loadEPKc ++ fun:_Z8tryShlibPKcb ++ fun:main ++} ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:malloc ++ fun:_dl_map_object ++ fun:openaux ++ fun:_dl_catch_error ++ fun:_dl_map_object_deps ++ fun:dl_open_worker ++ fun:_dl_catch_error ++ fun:_dl_open ++ fun:dlopen_doit ++ fun:_dl_catch_error ++ fun:_dlerror_run ++ fun:dlopen@@GLIBC_2.2.5 ++ fun:_ZN4qpid3sys5Shlib4loadEPKc ++ fun:_Z8tryShlibPKcb ++ fun:main ++} ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:malloc ++ fun:expand_dynamic_string_token ++ fun:_dl_map_object ++ fun:dl_open_worker ++ fun:_dl_catch_error ++ fun:_dl_open ++ fun:dlopen_doit ++ fun:_dl_catch_error ++ fun:_dlerror_run ++ fun:dlopen@@GLIBC_2.2.5 ++ fun:_ZN4qpid3sys5Shlib4loadEPKc ++ fun:_Z8tryShlibPKcb ++ fun:main ++} ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:malloc ++ fun:_dl_new_object ++ fun:_dl_map_object_from_fd ++ fun:_dl_map_object ++ fun:dl_open_worker ++ fun:_dl_catch_error ++ fun:_dl_open ++ fun:dlopen_doit ++ fun:_dl_catch_error ++ fun:_dlerror_run ++ fun:dlopen@@GLIBC_2.2.5 ++ fun:_ZN4qpid3sys5Shlib4loadEPKc ++ fun:_Z8tryShlibPKcb ++ fun:main ++} ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:calloc ++ fun:_dl_new_object ++ fun:_dl_map_object_from_fd ++ fun:_dl_map_object ++ fun:dl_open_worker ++ fun:_dl_catch_error ++ fun:_dl_open ++ fun:dlopen_doit ++ fun:_dl_catch_error ++ fun:_dlerror_run ++ fun:dlopen@@GLIBC_2.2.5 ++ fun:_ZN4qpid3sys5Shlib4loadEPKc ++ fun:_Z8tryShlibPKcb ++ fun:main ++} ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:malloc ++ fun:_dl_map_object_deps ++ fun:dl_open_worker ++ fun:_dl_catch_error ++ fun:_dl_open ++ fun:dlopen_doit ++ fun:_dl_catch_error ++ fun:_dlerror_run ++ fun:dlopen@@GLIBC_2.2.5 ++ fun:_ZN4qpid3sys5Shlib4loadEPKc ++ fun:_Z8tryShlibPKcb ++ fun:main ++} ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:calloc ++ fun:_dl_check_map_versions ++ fun:dl_open_worker ++ fun:_dl_catch_error ++ fun:_dl_open ++ fun:dlopen_doit ++ fun:_dl_catch_error ++ fun:_dlerror_run ++ fun:dlopen@@GLIBC_2.2.5 ++ fun:_ZN4qpid3sys5Shlib4loadEPKc ++ fun:_Z8tryShlibPKcb ++ fun:main ++} ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:_Znwm ++ fun:_ZN5boost15program_options19options_description3addERKS1_ ++ fun:main ++} + +Index: configure.ac +=================================================================== +--- configure.ac (revision 2179) ++++ configure.ac (working copy) +@@ -54,7 +54,6 @@ + # -Wshadow - warns about boost headers. + + if test "${enableval}" = yes; then +- gl_COMPILER_FLAGS(-Werror) + gl_COMPILER_FLAGS(-pedantic) + gl_COMPILER_FLAGS(-Wall) + gl_COMPILER_FLAGS(-Wextra) +@@ -63,7 +62,6 @@ + gl_COMPILER_FLAGS(-Wcast-qual) + gl_COMPILER_FLAGS(-Wcast-align) + gl_COMPILER_FLAGS(-Wno-long-long) +- gl_COMPILER_FLAGS(-Wvolatile-register-var) + gl_COMPILER_FLAGS(-Winvalid-pch) + gl_COMPILER_FLAGS(-Wno-system-headers) + AC_SUBST([WARNING_CFLAGS], [$COMPILER_FLAGS]) +Index: lib/BdbMessageStore.cpp +=================================================================== +--- lib/BdbMessageStore.cpp (revision 2179) ++++ lib/BdbMessageStore.cpp (working copy) +@@ -224,10 +224,6 @@ + try { + env.open(getBdbBaseDir().c_str(), DB_THREAD | DB_CREATE | DB_INIT_TXN | DB_INIT_LOCK | DB_INIT_LOG | DB_INIT_MPOOL | DB_USE_ENVIRON, 0); + } catch (const DbException& e) { +- if (e.get_errno() == DB_VERSION_MISMATCH) +- THROW_STORE_EXCEPTION_2("Database environment mismatch: This version of bd4 does not match that which created the store database. " +- "(If recovery is not important, delete the contents of the store directory. Otherwise, try upgrading the database using " +- "db_upgrade or using db_recover - but the db4-utils package must also be installed to use these utilities.)", e); + THROW_STORE_EXCEPTION_2("Error opening environment", e); + } + +Index: Makefile.am +=================================================================== +--- Makefile.am (revision 2179) ++++ Makefile.am (working copy) +@@ -1,4 +1,4 @@ +-AUTOMAKE_OPTIONS = 1.9.6 foreign ++AUTOMAKE_OPTIONS = 1.9.2 foreign + ACLOCAL_AMFLAGS = -I m4 + + EXTRA_DIST = README etc/rhmd.conf

15 years, 10 months

1
0
0 / 0

rhmessaging commits: r2240 - in store/trunk/cpp: rhel4-support and 1 other directory.

by rhmessaging-commits＠lists.jboss.org

Author: kpvdr Date: 2008-07-31 13:58:21 -0400 (Thu, 31 Jul 2008) New Revision: 2240 Added: store/trunk/cpp/rhel4-support/ store/trunk/cpp/rhel4-support/Makefile store/trunk/cpp/rhel4-support/README store/trunk/cpp/rhel4-support/rhel4.patch Log: Added dir for RHEL4 support Added: store/trunk/cpp/rhel4-support/Makefile =================================================================== --- store/trunk/cpp/rhel4-support/Makefile (rev 0) +++ store/trunk/cpp/rhel4-support/Makefile 2008-07-31 17:58:21 UTC (rev 2240) @@ -0,0 +1,7 @@ +.PHONY: default apply + +default: + @echo "Run 'make apply' to patch the source for RHEL4" + +apply: rhel4.patch + patch -d .. -p0 < rhel4.patch Added: store/trunk/cpp/rhel4-support/README =================================================================== --- store/trunk/cpp/rhel4-support/README (rev 0) +++ store/trunk/cpp/rhel4-support/README 2008-07-31 17:58:21 UTC (rev 2240) @@ -0,0 +1,8 @@ +To compile on RHEL4, run: + +make apply + +This will apply all necessary patches. Change dirs back to cpp +and run ./bootstrap and ./configure as ususal. Make sure that if +./configure is run with --with-qpid-checkout, it points to a version +of qpid which has also been patched for RHEL4. Added: store/trunk/cpp/rhel4-support/rhel4.patch =================================================================== --- store/trunk/cpp/rhel4-support/rhel4.patch (rev 0) +++ store/trunk/cpp/rhel4-support/rhel4.patch 2008-07-31 17:58:21 UTC (rev 2240) @@ -0,0 +1,276 @@ +Index: tests/.valgrind.supp +=================================================================== +--- tests/.valgrind.supp (revision 2179) ++++ tests/.valgrind.supp (working copy) +@@ -1,32 +1,205 @@ + { +- Benign error in libcpg. +- Memcheck:Param +- socketcall.sendmsg(msg.msg_iov[i]) +- obj:*/libpthread-2.5.so +- obj:*/libcpg.so.2.0.0 ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:_Znwm ++ fun:_ZNSs4_Rep9_S_createEmmRKSaIcE ++ obj:/usr/lib64/libstdc++.so.6.0.3 ++ fun:_ZNSsC1EPKcRKSaIcE ++ fun:_ZN4qpid34options_description_less_easy_initclEPKcPKN5boost15program_options14value_semanticES2_ ++ fun:_ZN4qpid3log7OptionsC1ERKSsS3_ ++ fun:_ZN4qpid3log6LoggerC1Ev + } +- + { +- Uninitialised value problem in _dl_relocate (F7, F8) +- Memcheck:Cond +- fun:_dl_relocate_object +- fun:*dl_* ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:_Znwm ++ fun:_ZNSt6vectorISsSaISsEE13_M_insert_auxEN9__gnu_cxx17__normal_iteratorIPSsS1_EERKSs ++ fun:_ZN4qpid7Options14register_namesESs ++ fun:_ZN4qpid34options_description_less_easy_initclEPKcPKN5boost15program_options14value_semanticES2_ ++ fun:_ZN4qpid3log7OptionsC1ERKSsS3_ ++ fun:_ZN4qpid3log6LoggerC1Ev + } +- + { +- False "possibly leaked" in boost program_options - global std::string var. ++ <insert a suppression name here> + Memcheck:Leak +- fun:_Znwj +- fun:_ZNSs4_Rep9_S_createEjjRKSaIcE +- obj:/usr/lib/libstdc++.so.6.0.8 +- fun:_ZNSsC1EPKcRKSaIcE +- obj:/usr/lib/libboost_program_options.so.1.33.1 ++ fun:_Znwm ++ fun:_ZN5boost15program_options29options_description_easy_initclEPKcPKNS0_14value_semanticES3_ ++ fun:_ZN4qpid34options_description_less_easy_initclEPKcPKN5boost15program_options14value_semanticES2_ ++ fun:_ZN4qpid3log7OptionsC1ERKSsS3_ ++ fun:_ZN4qpid3log6LoggerC1Ev + } ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:calloc ++ fun:_dl_allocate_tls ++ fun:pthread_create@@GLIBC_2.2.5 ++ fun:_ZN3rhm3jtt13jrnl_instance6run_tcEv ++ fun:_Z13constructor_1v ++ fun:_ZN5boost9unit_test9ut_detail17unit_test_monitor8functionEv ++ obj:/usr/lib64/libboost_unit_test_framework.so.1.32.0 ++ fun:_ZN5boost17execution_monitor7executeEbi ++ fun:_ZN5boost9unit_test9ut_detail17unit_test_monitor21execute_and_translateEPNS0_9test_caseEMS3_FvvEi ++ fun:_ZN5boost9unit_test9test_case3runEv ++ fun:_ZN5boost9unit_test10test_suite6do_runEv ++ fun:_ZN5boost9unit_test9test_case3runEv ++ fun:main ++} ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:_Znwm ++ fun:_ZN5boost9unit_test9test_caseC2ENS0_13basic_cstringIKcEEbmb ++} ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:calloc ++ fun:_dl_allocate_tls ++ fun:pthread_create@@GLIBC_2.2.5 ++ fun:_ZN4qpid6broker5Timer5startEv ++ fun:_ZN4qpid6broker5TimerC1Ev ++} + + { +- Bogus epoll_ctl error on i386 +- Memcheck:Param +- epoll_ctl(event) +- fun:epoll_ctl ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:malloc ++ fun:decompose_rpath ++ fun:_dl_map_object ++ fun:openaux ++ fun:_dl_catch_error ++ fun:_dl_map_object_deps ++ fun:dl_open_worker ++ fun:_dl_catch_error ++ fun:_dl_open ++ fun:dlopen_doit ++ fun:_dl_catch_error ++ fun:_dlerror_run ++ fun:dlopen@@GLIBC_2.2.5 ++ fun:_ZN4qpid3sys5Shlib4loadEPKc ++ fun:_Z8tryShlibPKcb ++ fun:main + } ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:calloc ++ fun:_dlerror_run ++ fun:dlopen@@GLIBC_2.2.5 ++ fun:_ZN4qpid3sys5Shlib4loadEPKc ++ fun:_Z8tryShlibPKcb ++ fun:main ++} ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:malloc ++ fun:_dl_map_object ++ fun:openaux ++ fun:_dl_catch_error ++ fun:_dl_map_object_deps ++ fun:dl_open_worker ++ fun:_dl_catch_error ++ fun:_dl_open ++ fun:dlopen_doit ++ fun:_dl_catch_error ++ fun:_dlerror_run ++ fun:dlopen@@GLIBC_2.2.5 ++ fun:_ZN4qpid3sys5Shlib4loadEPKc ++ fun:_Z8tryShlibPKcb ++ fun:main ++} ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:malloc ++ fun:expand_dynamic_string_token ++ fun:_dl_map_object ++ fun:dl_open_worker ++ fun:_dl_catch_error ++ fun:_dl_open ++ fun:dlopen_doit ++ fun:_dl_catch_error ++ fun:_dlerror_run ++ fun:dlopen@@GLIBC_2.2.5 ++ fun:_ZN4qpid3sys5Shlib4loadEPKc ++ fun:_Z8tryShlibPKcb ++ fun:main ++} ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:malloc ++ fun:_dl_new_object ++ fun:_dl_map_object_from_fd ++ fun:_dl_map_object ++ fun:dl_open_worker ++ fun:_dl_catch_error ++ fun:_dl_open ++ fun:dlopen_doit ++ fun:_dl_catch_error ++ fun:_dlerror_run ++ fun:dlopen@@GLIBC_2.2.5 ++ fun:_ZN4qpid3sys5Shlib4loadEPKc ++ fun:_Z8tryShlibPKcb ++ fun:main ++} ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:calloc ++ fun:_dl_new_object ++ fun:_dl_map_object_from_fd ++ fun:_dl_map_object ++ fun:dl_open_worker ++ fun:_dl_catch_error ++ fun:_dl_open ++ fun:dlopen_doit ++ fun:_dl_catch_error ++ fun:_dlerror_run ++ fun:dlopen@@GLIBC_2.2.5 ++ fun:_ZN4qpid3sys5Shlib4loadEPKc ++ fun:_Z8tryShlibPKcb ++ fun:main ++} ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:malloc ++ fun:_dl_map_object_deps ++ fun:dl_open_worker ++ fun:_dl_catch_error ++ fun:_dl_open ++ fun:dlopen_doit ++ fun:_dl_catch_error ++ fun:_dlerror_run ++ fun:dlopen@@GLIBC_2.2.5 ++ fun:_ZN4qpid3sys5Shlib4loadEPKc ++ fun:_Z8tryShlibPKcb ++ fun:main ++} ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:calloc ++ fun:_dl_check_map_versions ++ fun:dl_open_worker ++ fun:_dl_catch_error ++ fun:_dl_open ++ fun:dlopen_doit ++ fun:_dl_catch_error ++ fun:_dlerror_run ++ fun:dlopen@@GLIBC_2.2.5 ++ fun:_ZN4qpid3sys5Shlib4loadEPKc ++ fun:_Z8tryShlibPKcb ++ fun:main ++} ++{ ++ <insert a suppression name here> ++ Memcheck:Leak ++ fun:_Znwm ++ fun:_ZN5boost15program_options19options_description3addERKS1_ ++ fun:main ++} + +Index: configure.ac +=================================================================== +--- configure.ac (revision 2179) ++++ configure.ac (working copy) +@@ -54,7 +54,6 @@ + # -Wshadow - warns about boost headers. + + if test "${enableval}" = yes; then +- gl_COMPILER_FLAGS(-Werror) + gl_COMPILER_FLAGS(-pedantic) + gl_COMPILER_FLAGS(-Wall) + gl_COMPILER_FLAGS(-Wextra) +@@ -63,7 +62,6 @@ + gl_COMPILER_FLAGS(-Wcast-qual) + gl_COMPILER_FLAGS(-Wcast-align) + gl_COMPILER_FLAGS(-Wno-long-long) +- gl_COMPILER_FLAGS(-Wvolatile-register-var) + gl_COMPILER_FLAGS(-Winvalid-pch) + gl_COMPILER_FLAGS(-Wno-system-headers) + AC_SUBST([WARNING_CFLAGS], [$COMPILER_FLAGS]) +Index: lib/BdbMessageStore.cpp +=================================================================== +--- lib/BdbMessageStore.cpp (revision 2179) ++++ lib/BdbMessageStore.cpp (working copy) +@@ -230,10 +230,6 @@ + try { + env.open(getBdbBaseDir().c_str(), DB_THREAD | DB_CREATE | DB_INIT_TXN | DB_INIT_LOCK | DB_INIT_LOG | DB_INIT_MPOOL | DB_USE_ENVIRON, 0); + } catch (const DbException& e) { +- if (e.get_errno() == DB_VERSION_MISMATCH) +- THROW_STORE_EXCEPTION_2("Database environment mismatch: This version of bd4 does not match that which created the store database. " +- "(If recovery is not important, delete the contents of the store directory. Otherwise, try upgrading the database using " +- "db_upgrade or using db_recover - but the db4-utils package must also be installed to use these utilities.)", e); + THROW_STORE_EXCEPTION_2("Error opening environment", e); + } + +Index: Makefile.am +=================================================================== +--- Makefile.am (revision 2179) ++++ Makefile.am (working copy) +@@ -1,4 +1,4 @@ +-AUTOMAKE_OPTIONS = 1.9.6 foreign ++AUTOMAKE_OPTIONS = 1.9.2 foreign + ACLOCAL_AMFLAGS = -I m4 + + EXTRA_DIST = README etc/rhmd.conf

15 years, 10 months

1
0
0 / 0

rhmessaging commits: r2239 - store/trunk/cpp/lib/gen/qpid/management.

by rhmessaging-commits＠lists.jboss.org

Author: tedross Date: 2008-07-31 11:51:01 -0400 (Thu, 31 Jul 2008) New Revision: 2239 Modified: store/trunk/cpp/lib/gen/qpid/management/Journal.h store/trunk/cpp/lib/gen/qpid/management/Store.h Log: regenerated management files Modified: store/trunk/cpp/lib/gen/qpid/management/Journal.h =================================================================== --- store/trunk/cpp/lib/gen/qpid/management/Journal.h 2008-07-30 20:28:39 UTC (rev 2238) +++ store/trunk/cpp/lib/gen/qpid/management/Journal.h 2008-07-31 15:51:01 UTC (rev 2239) @@ -134,9 +134,9 @@ void setReference(uint64_t objectId) { queueRef = objectId; } - std::string getPackageName (void) { return packageName; } - std::string getClassName (void) { return className; } - uint8_t* getMd5Sum (void) { return md5Sum; } + std::string& getPackageName (void) { return packageName; } + std::string& getClassName (void) { return className; } + uint8_t* getMd5Sum (void) { return md5Sum; } // Method IDs static const uint32_t METHOD_EXPAND = 1; Modified: store/trunk/cpp/lib/gen/qpid/management/Store.h =================================================================== --- store/trunk/cpp/lib/gen/qpid/management/Store.h 2008-07-30 20:28:39 UTC (rev 2238) +++ store/trunk/cpp/lib/gen/qpid/management/Store.h 2008-07-31 15:51:01 UTC (rev 2239) @@ -107,9 +107,9 @@ - std::string getPackageName (void) { return packageName; } - std::string getClassName (void) { return className; } - uint8_t* getMd5Sum (void) { return md5Sum; } + std::string& getPackageName (void) { return packageName; } + std::string& getClassName (void) { return className; } + uint8_t* getMd5Sum (void) { return md5Sum; } // Method IDs

15 years, 10 months

1
0
0 / 0

rhmessaging commits: r2238 - store/branches/mrg-1.0/cpp/lib.

by rhmessaging-commits＠lists.jboss.org

Author: kpvdr Date: 2008-07-30 16:28:39 -0400 (Wed, 30 Jul 2008) New Revision: 2238 Modified: store/branches/mrg-1.0/cpp/lib/BdbMessageStore.cpp store/branches/mrg-1.0/cpp/lib/BdbMessageStore.h Log: Backports of trunk r.2236,2237: Improvements on fix for BZ457239, also tightening up on exception handling around transactions in BdbMessageStore. Also added txn to one addtional call omitted on previouis fixes. Modified: store/branches/mrg-1.0/cpp/lib/BdbMessageStore.cpp =================================================================== --- store/branches/mrg-1.0/cpp/lib/BdbMessageStore.cpp 2008-07-30 20:09:46 UTC (rev 2237) +++ store/branches/mrg-1.0/cpp/lib/BdbMessageStore.cpp 2008-07-30 20:28:39 UTC (rev 2238) @@ -249,7 +249,7 @@ } catch (const DbException& e) { txn.abort(); THROW_STORE_EXCEPTION_2("Error opening databases", e); - } catch (const std::exception& e) { + } catch (...) { txn.abort(); throw; } @@ -359,7 +359,6 @@ THROW_STORE_EXCEPTION(std::string("Queue ") + queue.getName() + ": create() failed: " + e.what()); } try { - qpid::sys::Mutex::ScopedLock sl(bdbQueueLock); if (!create(queueDb, queueIdSequence, queue)) { THROW_STORE_EXCEPTION("Queue already exists: " + queue.getName()); } @@ -435,7 +434,16 @@ Dbt key(&id, sizeof(id)); BufferValue value (p); - int status = db.put(0, &key, &value, DB_NOOVERWRITE | DB_AUTO_COMMIT); + int status; + TxnCtxt txn; + txn.begin(env, true); + try { + status = db.put(txn.get(), &key, &value, DB_NOOVERWRITE); + txn.commit(); + } catch (...) { + txn.abort(); + throw; + } if (status == DB_KEYEXIST) { return false; } else { @@ -461,8 +469,13 @@ BindingDbt value(e, q, k, a); TxnCtxt txn; txn.begin(env, true); - put(bindingDb, txn.get(), key, value); - txn.commit(); + try { + put(bindingDb, txn.get(), key, value); + txn.commit(); + } catch (...) { + txn.abort(); + throw; + } } void BdbMessageStore::unbind(const PersistableExchange& e, @@ -498,10 +511,12 @@ recoverGeneral(txn, registry); txn.commit(); - } catch (const DbException& e) { txn.abort(); THROW_STORE_EXCEPTION_2("Error on recovery", e); + } catch (...) { + txn.abort(); + throw; } //recover transactions: @@ -785,7 +800,7 @@ value.buffer.record(); TxnCtxt txn; - txn.begin(env); + txn.begin(env, true); try { if (messageDb.get(txn.get(), &key, &value, 0) == DB_NOTFOUND) { txn.abort(); @@ -804,6 +819,9 @@ } catch (const DbException& e) { txn.abort(); THROW_STORE_EXCEPTION("Unexpected BDB error in BdbMessageStore::getExternMessage(): " + string(e.what())); + } catch (...) { + txn.abort(); + throw; } return ret; } @@ -922,7 +940,7 @@ store(NULL, &txn, key, msg, true); msg->setPersistenceId(messageId); txn.commit(); - } catch (const std::exception& e) { + } catch (...) { txn.abort(); throw; } @@ -943,6 +961,9 @@ } catch (const DbException& e) { txn.abort(); THROW_STORE_EXCEPTION_2("Error destroying message", e); + } catch (...) { + txn.abort(); + throw; } } } @@ -950,7 +971,17 @@ u_int64_t BdbMessageStore::getRecordSize(Db& db, Dbt& key) { - return getRecordSize(0, db, key); + u_int64_t ret = 0; + TxnCtxt txn; + txn.begin(env, true); + try { + ret = getRecordSize(txn.get(), db, key); + txn.commit(); + } catch (...) { + txn.abort(); + throw; + } + return ret; } u_int64_t BdbMessageStore::getRecordSize(DbTxn* txn, @@ -977,6 +1008,8 @@ checkInit(); u_int64_t messageId (msg->getPersistenceId()); if (messageId != 0) { + TxnCtxt txn; + txn.begin(env, true); try { Dbt key (&messageId, sizeof(messageId)); u_int64_t offset = getRecordSize(messageDb, key); @@ -992,9 +1025,14 @@ value.set_flags(DB_DBT_USERMEM | DB_DBT_PARTIAL); value.set_doff(offset); value.set_dlen(size); - messageDb.put(0, &key, &value, DB_AUTO_COMMIT); + messageDb.put(txn.get(), &key, &value, 0); + txn.commit(); } catch (const DbException& e) { + txn.abort(); THROW_STORE_EXCEPTION_2("Error appending content", e); + } catch (...) { + txn.abort(); + throw; } } else { THROW_STORE_EXCEPTION("Cannot append content. Message not known to store!"); @@ -1035,16 +1073,18 @@ value.set_dlen(length); int status = messageDb.get(txn.get(), &key, &value, 0); if (status == DB_NOTFOUND) { - txn.abort(); delete [] buffer; THROW_STORE_EXCEPTION("Cannot load content. Message not known to store!"); - } else { - txn.commit(); - data.assign(buffer, value.get_size()); - delete [] buffer; } + data.assign(buffer, value.get_size()); + delete [] buffer; + txn.commit(); } catch (const DbException& e) { + txn.abort(); THROW_STORE_EXCEPTION_2("Error loading content", e); + } catch (...) { + txn.abort(); + throw; } } else { THROW_STORE_EXCEPTION("Cannot load content. Message not known to store!"); @@ -1378,26 +1418,32 @@ TxnCtxt txn; txn.begin(env, true); try { - Cursor bindings; - bindings.open(bindingDb, txn.get()); + { + Cursor bindings; + bindings.open(bindingDb, txn.get()); - IdDbt key; - Dbt value; - while (bindings.next(key, value)) { - Buffer buffer(reinterpret_cast<char*>(value.get_data()), value.get_size()); - if (buffer.available() < 8) { - THROW_STORE_EXCEPTION("Not enough data for binding"); + IdDbt key; + Dbt value; + while (bindings.next(key, value)) { + Buffer buffer(reinterpret_cast<char*>(value.get_data()), value.get_size()); + if (buffer.available() < 8) { + THROW_STORE_EXCEPTION("Not enough data for binding"); + } + uint64_t queueId = buffer.getLongLong(); + if (queue.getPersistenceId() == queueId) { + bindings->del(0); + QPID_LOG(debug, "Deleting binding for " << queue.getName() << " " << key.id << "->" << queueId); + } } - uint64_t queueId = buffer.getLongLong(); - if (queue.getPersistenceId() == queueId) { - bindings->del(0); - QPID_LOG(debug, "Deleting binding for " << queue.getName() << " " << key.id << "->" << queueId); - } } + txn.commit(); } catch (const std::exception& e) { + txn.abort(); THROW_STORE_EXCEPTION_2("Error deleting bindings", e.what()); + } catch (...) { + txn.abort(); + throw; } - txn.commit(); QPID_LOG(debug, "Deleted all bindings for " << queue.getName() << ":" << queue.getPersistenceId()); } @@ -1408,33 +1454,39 @@ TxnCtxt txn; txn.begin(env, true); try { - Cursor bindings; - bindings.open(bindingDb, txn.get()); + { + Cursor bindings; + bindings.open(bindingDb, txn.get()); - IdDbt key(exchange.getPersistenceId()); - Dbt value; + IdDbt key(exchange.getPersistenceId()); + Dbt value; - for (int status = bindings->get(&key, &value, DB_SET); status == 0; status = bindings->get(&key, &value, DB_NEXT_DUP)) { - Buffer buffer(reinterpret_cast<char*>(value.get_data()), value.get_size()); - if (buffer.available() < 8) { - THROW_STORE_EXCEPTION("Not enough data for binding"); - } - uint64_t queueId = buffer.getLongLong(); - if (queue.getPersistenceId() == queueId) { - std::string q; - std::string k; - buffer.getShortString(q); - buffer.getShortString(k); - if (bkey == k) { - bindings->del(0); - QPID_LOG(debug, "Deleting binding for " << queue.getName() << " " << key.id << "->" << queueId); + for (int status = bindings->get(&key, &value, DB_SET); status == 0; status = bindings->get(&key, &value, DB_NEXT_DUP)) { + Buffer buffer(reinterpret_cast<char*>(value.get_data()), value.get_size()); + if (buffer.available() < 8) { + THROW_STORE_EXCEPTION("Not enough data for binding"); } + uint64_t queueId = buffer.getLongLong(); + if (queue.getPersistenceId() == queueId) { + std::string q; + std::string k; + buffer.getShortString(q); + buffer.getShortString(k); + if (bkey == k) { + bindings->del(0); + QPID_LOG(debug, "Deleting binding for " << queue.getName() << " " << key.id << "->" << queueId); + } + } } } + txn.commit(); } catch (const std::exception& e) { + txn.abort(); THROW_STORE_EXCEPTION_2("Error deleting bindings", e.what()); + } catch (...) { + txn.abort(); + throw; } - txn.commit(); } string BdbMessageStore::getJrnlBaseDir() Modified: store/branches/mrg-1.0/cpp/lib/BdbMessageStore.h =================================================================== --- store/branches/mrg-1.0/cpp/lib/BdbMessageStore.h 2008-07-30 20:09:46 UTC (rev 2237) +++ store/branches/mrg-1.0/cpp/lib/BdbMessageStore.h 2008-07-30 20:28:39 UTC (rev 2238) @@ -115,7 +115,6 @@ static qpid::sys::Duration defJournalFlushTimeout; qpid::management::Store::shared_ptr mgmtObject; qpid::sys::Mutex jrnlCreateLock; - qpid::sys::Mutex bdbQueueLock; // Parameter validation and calculation static u_int16_t chkJrnlNumFilesParam(const u_int16_t param,

15 years, 10 months

1
0
0 / 0

rhmessaging commits: r2237 - store/trunk/cpp/lib.

by rhmessaging-commits＠lists.jboss.org

Author: kpvdr Date: 2008-07-30 16:09:46 -0400 (Wed, 30 Jul 2008) New Revision: 2237 Modified: store/trunk/cpp/lib/BdbMessageStore.cpp Log: Fix for BDB txn error created in last checkin, also added txn to one addtional call omitted on previouis fixes. Modified: store/trunk/cpp/lib/BdbMessageStore.cpp =================================================================== --- store/trunk/cpp/lib/BdbMessageStore.cpp 2008-07-30 18:32:51 UTC (rev 2236) +++ store/trunk/cpp/lib/BdbMessageStore.cpp 2008-07-30 20:09:46 UTC (rev 2237) @@ -452,7 +452,7 @@ TxnCtxt txn; txn.begin(env, true); try { - status = db.put(txn.get(), &key, &value, DB_NOOVERWRITE | DB_AUTO_COMMIT); + status = db.put(txn.get(), &key, &value, DB_NOOVERWRITE); txn.commit(); } catch (...) { txn.abort(); @@ -1035,6 +1035,8 @@ checkInit(); u_int64_t messageId (msg->getPersistenceId()); if (messageId != 0) { + TxnCtxt txn; + txn.begin(env, true); try { Dbt key (&messageId, sizeof(messageId)); u_int64_t offset = getRecordSize(messageDb, key); @@ -1050,9 +1052,14 @@ value.set_flags(DB_DBT_USERMEM | DB_DBT_PARTIAL); value.set_doff(offset); value.set_dlen(size); - messageDb.put(0, &key, &value, DB_AUTO_COMMIT); + messageDb.put(txn.get(), &key, &value, 0); + txn.commit(); } catch (const DbException& e) { + txn.abort(); THROW_STORE_EXCEPTION_2("Error appending content", e); + } catch (...) { + txn.abort(); + throw; } } else { THROW_STORE_EXCEPTION("Cannot append content. Message not known to store!");

15 years, 10 months

1
0
0 / 0

rhmessaging commits: r2236 - store/trunk/cpp/lib.

by rhmessaging-commits＠lists.jboss.org

Author: kpvdr Date: 2008-07-30 14:32:51 -0400 (Wed, 30 Jul 2008) New Revision: 2236 Modified: store/trunk/cpp/lib/BdbMessageStore.cpp store/trunk/cpp/lib/BdbMessageStore.h Log: Improvements on fix for BZ457239, also tightening up on exception handling around transactions in BdbMessageStore Modified: store/trunk/cpp/lib/BdbMessageStore.cpp =================================================================== --- store/trunk/cpp/lib/BdbMessageStore.cpp 2008-07-30 16:35:25 UTC (rev 2235) +++ store/trunk/cpp/lib/BdbMessageStore.cpp 2008-07-30 18:32:51 UTC (rev 2236) @@ -255,7 +255,7 @@ } catch (const DbException& e) { txn.abort(); THROW_STORE_EXCEPTION_2("Error opening databases", e); - } catch (const std::exception& e) { + } catch (...) { txn.abort(); throw; } @@ -373,7 +373,6 @@ THROW_STORE_EXCEPTION(std::string("Queue ") + queue.getName() + ": create() failed: " + e.what()); } try { - qpid::sys::Mutex::ScopedLock sl(bdbQueueLock); if (!create(queueDb, queueIdSequence, queue)) { THROW_STORE_EXCEPTION("Queue already exists: " + queue.getName()); } @@ -449,7 +448,16 @@ Dbt key(&id, sizeof(id)); BufferValue value (p); - int status = db.put(0, &key, &value, DB_NOOVERWRITE | DB_AUTO_COMMIT); + int status; + TxnCtxt txn; + txn.begin(env, true); + try { + status = db.put(txn.get(), &key, &value, DB_NOOVERWRITE | DB_AUTO_COMMIT); + txn.commit(); + } catch (...) { + txn.abort(); + throw; + } if (status == DB_KEYEXIST) { return false; } else { @@ -475,8 +483,13 @@ BindingDbt value(e, q, k, a); TxnCtxt txn; txn.begin(env, true); - put(bindingDb, txn.get(), key, value); - txn.commit(); + try { + put(bindingDb, txn.get(), key, value); + txn.commit(); + } catch (...) { + txn.abort(); + throw; + } } void BdbMessageStore::unbind(const PersistableExchange& e, @@ -512,10 +525,12 @@ recoverGeneral(txn, registry); txn.commit(); - } catch (const DbException& e) { txn.abort(); THROW_STORE_EXCEPTION_2("Error on recovery", e); + } catch (...) { + txn.abort(); + throw; } //recover transactions: @@ -811,7 +826,7 @@ value.buffer.record(); TxnCtxt txn; - txn.begin(env); + txn.begin(env, true); try { if (messageDb.get(txn.get(), &key, &value, 0) == DB_NOTFOUND) { txn.abort(); @@ -831,6 +846,9 @@ } catch (const DbException& e) { txn.abort(); THROW_STORE_EXCEPTION("Unexpected BDB error in BdbMessageStore::getExternMessage(): " + string(e.what())); + } catch (...) { + txn.abort(); + throw; } return ret; } @@ -949,7 +967,7 @@ store(NULL, &txn, key, msg, true); msg->setPersistenceId(messageId); txn.commit(); - } catch (const std::exception& e) { + } catch (...) { txn.abort(); throw; } @@ -970,6 +988,9 @@ } catch (const DbException& e) { txn.abort(); THROW_STORE_EXCEPTION_2("Error destroying message", e); + } catch (...) { + txn.abort(); + throw; } } } @@ -979,11 +1000,11 @@ { u_int64_t ret = 0; TxnCtxt txn; - txn.begin(env); + txn.begin(env, true); try { ret = getRecordSize(txn.get(), db, key); txn.commit(); - } catch (const std::exception& e) { + } catch (...) { txn.abort(); throw; } @@ -1072,16 +1093,18 @@ value.set_dlen(length); int status = messageDb.get(txn.get(), &key, &value, 0); if (status == DB_NOTFOUND) { - txn.abort(); delete [] buffer; THROW_STORE_EXCEPTION("Cannot load content. Message not known to store!"); - } else { - txn.commit(); - data.assign(buffer, value.get_size()); - delete [] buffer; } + data.assign(buffer, value.get_size()); + delete [] buffer; + txn.commit(); } catch (const DbException& e) { + txn.abort(); THROW_STORE_EXCEPTION_2("Error loading content", e); + } catch (...) { + txn.abort(); + throw; } } else { THROW_STORE_EXCEPTION("Cannot load content. Message not known to store!"); @@ -1426,26 +1449,32 @@ TxnCtxt txn; txn.begin(env, true); try { - Cursor bindings; - bindings.open(bindingDb, txn.get()); + { + Cursor bindings; + bindings.open(bindingDb, txn.get()); - IdDbt key; - Dbt value; - while (bindings.next(key, value)) { - Buffer buffer(reinterpret_cast<char*>(value.get_data()), value.get_size()); - if (buffer.available() < 8) { - THROW_STORE_EXCEPTION("Not enough data for binding"); + IdDbt key; + Dbt value; + while (bindings.next(key, value)) { + Buffer buffer(reinterpret_cast<char*>(value.get_data()), value.get_size()); + if (buffer.available() < 8) { + THROW_STORE_EXCEPTION("Not enough data for binding"); + } + uint64_t queueId = buffer.getLongLong(); + if (queue.getPersistenceId() == queueId) { + bindings->del(0); + QPID_LOG(debug, "Deleting binding for " << queue.getName() << " " << key.id << "->" << queueId); + } } - uint64_t queueId = buffer.getLongLong(); - if (queue.getPersistenceId() == queueId) { - bindings->del(0); - QPID_LOG(debug, "Deleting binding for " << queue.getName() << " " << key.id << "->" << queueId); - } } + txn.commit(); } catch (const std::exception& e) { + txn.abort(); THROW_STORE_EXCEPTION_2("Error deleting bindings", e.what()); + } catch (...) { + txn.abort(); + throw; } - txn.commit(); QPID_LOG(debug, "Deleted all bindings for " << queue.getName() << ":" << queue.getPersistenceId()); } @@ -1456,33 +1485,39 @@ TxnCtxt txn; txn.begin(env, true); try { - Cursor bindings; - bindings.open(bindingDb, txn.get()); + { + Cursor bindings; + bindings.open(bindingDb, txn.get()); - IdDbt key(exchange.getPersistenceId()); - Dbt value; + IdDbt key(exchange.getPersistenceId()); + Dbt value; - for (int status = bindings->get(&key, &value, DB_SET); status == 0; status = bindings->get(&key, &value, DB_NEXT_DUP)) { - Buffer buffer(reinterpret_cast<char*>(value.get_data()), value.get_size()); - if (buffer.available() < 8) { - THROW_STORE_EXCEPTION("Not enough data for binding"); - } - uint64_t queueId = buffer.getLongLong(); - if (queue.getPersistenceId() == queueId) { - std::string q; - std::string k; - buffer.getShortString(q); - buffer.getShortString(k); - if (bkey == k) { - bindings->del(0); - QPID_LOG(debug, "Deleting binding for " << queue.getName() << " " << key.id << "->" << queueId); + for (int status = bindings->get(&key, &value, DB_SET); status == 0; status = bindings->get(&key, &value, DB_NEXT_DUP)) { + Buffer buffer(reinterpret_cast<char*>(value.get_data()), value.get_size()); + if (buffer.available() < 8) { + THROW_STORE_EXCEPTION("Not enough data for binding"); } + uint64_t queueId = buffer.getLongLong(); + if (queue.getPersistenceId() == queueId) { + std::string q; + std::string k; + buffer.getShortString(q); + buffer.getShortString(k); + if (bkey == k) { + bindings->del(0); + QPID_LOG(debug, "Deleting binding for " << queue.getName() << " " << key.id << "->" << queueId); + } + } } } + txn.commit(); } catch (const std::exception& e) { + txn.abort(); THROW_STORE_EXCEPTION_2("Error deleting bindings", e.what()); + } catch (...) { + txn.abort(); + throw; } - txn.commit(); } string BdbMessageStore::getJrnlBaseDir() Modified: store/trunk/cpp/lib/BdbMessageStore.h =================================================================== --- store/trunk/cpp/lib/BdbMessageStore.h 2008-07-30 16:35:25 UTC (rev 2235) +++ store/trunk/cpp/lib/BdbMessageStore.h 2008-07-30 18:32:51 UTC (rev 2236) @@ -115,7 +115,6 @@ static qpid::sys::Duration defJournalFlushTimeout; qpid::management::Store* mgmtObject; qpid::sys::Mutex jrnlCreateLock; - qpid::sys::Mutex bdbQueueLock; // Parameter validation and calculation static u_int16_t chkJrnlNumFilesParam(const u_int16_t param,

15 years, 10 months

1
0
0 / 0

rhmessaging commits: r2235 - mgmt/trunk/mint/python/mint.

by rhmessaging-commits＠lists.jboss.org

Author: justi9 Date: 2008-07-30 12:35:25 -0400 (Wed, 30 Jul 2008) New Revision: 2235 Modified: mgmt/trunk/mint/python/mint/update.py Log: Also catch deletes on prop updates, not only stat updates Modified: mgmt/trunk/mint/python/mint/update.py =================================================================== --- mgmt/trunk/mint/python/mint/update.py 2008-07-30 15:20:43 UTC (rev 2234) +++ mgmt/trunk/mint/python/mint/update.py 2008-07-30 16:35:25 UTC (rev 2235) @@ -156,11 +156,17 @@ attrs["creationTime"] = datetime.fromtimestamp \ (self.timestamps[1]/1000000000) + if self.timestamps[2] != 0: + attrs["deletionTime"] = datetime.fromtimestamp \ + (self.timestamps[2]/1000000000) + try: obj = self.conn.getObject(cls, id) except mint.ObjectNotFound: obj = cls() + log.debug("%s(%i) created", cls.__name__, obj.id) + #obj.sourceId = id #obj.sourceBrokerId = self.conn.brokerId @@ -171,6 +177,9 @@ obj.set(**attrs) obj.syncUpdate() + if obj.deletionTime: + log.debug("%s(%i) marked deleted", cls.__name__, obj.id) + # XXX refactor this to take advantage of the get/create logic # above if isinstance(obj, mint.Broker) and obj.managedBroker: @@ -225,6 +234,8 @@ statsobj.set(**attrs) statsobj.syncUpdate() + # XXX not sure if this should happen here. makes more sense in + # prop update if self.timestamps[2] != 0: obj.deletionTime = datetime.fromtimestamp(self.timestamps[2]/1000000000)

15 years, 10 months

1
0
0 / 0

rhmessaging commits: r2234 - store/branches/mrg-1.0/cpp/lib.

by rhmessaging-commits＠lists.jboss.org

Author: kpvdr Date: 2008-07-30 11:20:43 -0400 (Wed, 30 Jul 2008) New Revision: 2234 Modified: store/branches/mrg-1.0/cpp/lib/BdbMessageStore.cpp store/branches/mrg-1.0/cpp/lib/BdbMessageStore.h Log: Backport of trunk r.2233: Proposed fix for BZ457239: "Qpidd sometimes deadlocks on queue_declare from Python". BDB is not always thread safe for certain operations. Added a lock on queue create (the cause of this bug), added local transaction to BdbMessageStore::getRecordSize() (used by BdbMessageStore::appendContent()) and BdbMessageStore::getExternMessage() which were found by inspection to be vulnerable for similar reasons. Modified: store/branches/mrg-1.0/cpp/lib/BdbMessageStore.cpp =================================================================== --- store/branches/mrg-1.0/cpp/lib/BdbMessageStore.cpp 2008-07-30 15:00:18 UTC (rev 2233) +++ store/branches/mrg-1.0/cpp/lib/BdbMessageStore.cpp 2008-07-30 15:20:43 UTC (rev 2234) @@ -359,6 +359,7 @@ THROW_STORE_EXCEPTION(std::string("Queue ") + queue.getName() + ": create() failed: " + e.what()); } try { + qpid::sys::Mutex::ScopedLock sl(bdbQueueLock); if (!create(queueDb, queueIdSequence, queue)) { THROW_STORE_EXCEPTION("Queue already exists: " + queue.getName()); } @@ -776,23 +777,35 @@ uint64_t messageId, unsigned& headerSize) { + RecoverableMessage::shared_ptr ret; Dbt key (&messageId, sizeof(messageId)); size_t preamble_length = sizeof(u_int32_t); /*header size*/ BufferValue value(preamble_length, 0); value.buffer.record(); - if (messageDb.get(0, &key, &value, 0) == DB_NOTFOUND) { - THROW_STORE_EXCEPTION("Cannot load content. Message not known to store!"); - } - //read header only to begin with - headerSize = value.buffer.getLong(); - BufferValue header(headerSize, preamble_length); - if (messageDb.get(0, &key, &header, 0) == DB_NOTFOUND) { - THROW_STORE_EXCEPTION("Cannot load content. Message not known to store!"); + TxnCtxt txn; + txn.begin(env); + try { + if (messageDb.get(txn.get(), &key, &value, 0) == DB_NOTFOUND) { + txn.abort(); + THROW_STORE_EXCEPTION("Cannot load content. Message not known to store!"); + } + //read header only to begin with + headerSize = value.buffer.getLong(); + + BufferValue header(headerSize, preamble_length); + if (messageDb.get(txn.get(), &key, &header, 0) == DB_NOTFOUND) { + txn.abort(); + THROW_STORE_EXCEPTION("Cannot load content. Message not known to store!"); + } + ret = recovery.recoverMessage(header.buffer); + txn.commit(); + } catch (const DbException& e) { + txn.abort(); + THROW_STORE_EXCEPTION("Unexpected BDB error in BdbMessageStore::getExternMessage(): " + string(e.what())); } - - return recovery.recoverMessage(header.buffer); + return ret; } int BdbMessageStore::enqueueMessage(TxnCtxt& txn, Modified: store/branches/mrg-1.0/cpp/lib/BdbMessageStore.h =================================================================== --- store/branches/mrg-1.0/cpp/lib/BdbMessageStore.h 2008-07-30 15:00:18 UTC (rev 2233) +++ store/branches/mrg-1.0/cpp/lib/BdbMessageStore.h 2008-07-30 15:20:43 UTC (rev 2234) @@ -115,6 +115,7 @@ static qpid::sys::Duration defJournalFlushTimeout; qpid::management::Store::shared_ptr mgmtObject; qpid::sys::Mutex jrnlCreateLock; + qpid::sys::Mutex bdbQueueLock; // Parameter validation and calculation static u_int16_t chkJrnlNumFilesParam(const u_int16_t param,

15 years, 10 months

1
0
0 / 0

rhmessaging commits: r2233 - store/trunk/cpp/lib.

by rhmessaging-commits＠lists.jboss.org

Author: kpvdr Date: 2008-07-30 11:00:18 -0400 (Wed, 30 Jul 2008) New Revision: 2233 Modified: store/trunk/cpp/lib/BdbMessageStore.cpp store/trunk/cpp/lib/BdbMessageStore.h Log: Proposed fix for BZ457239: "Qpidd sometimes deadlocks on queue_declare from Python". BDB is not always thread safe for certain operations. Added a lock on queue create (the cause of this bug), added local transaction to BdbMessageStore::getRecordSize() (used by BdbMessageStore::appendContent()) and BdbMessageStore::getExternMessage() which were found by inspection to be vulnerable for similar reasons. Modified: store/trunk/cpp/lib/BdbMessageStore.cpp =================================================================== --- store/trunk/cpp/lib/BdbMessageStore.cpp 2008-07-29 20:13:17 UTC (rev 2232) +++ store/trunk/cpp/lib/BdbMessageStore.cpp 2008-07-30 15:00:18 UTC (rev 2233) @@ -373,6 +373,7 @@ THROW_STORE_EXCEPTION(std::string("Queue ") + queue.getName() + ": create() failed: " + e.what()); } try { + qpid::sys::Mutex::ScopedLock sl(bdbQueueLock); if (!create(queueDb, queueIdSequence, queue)) { THROW_STORE_EXCEPTION("Queue already exists: " + queue.getName()); } @@ -802,23 +803,36 @@ uint64_t messageId, unsigned& headerSize) { + RecoverableMessage::shared_ptr ret; Dbt key (&messageId, sizeof(messageId)); size_t preamble_length = sizeof(u_int32_t); /*header size*/ BufferValue value(preamble_length, 0); value.buffer.record(); - if (messageDb.get(0, &key, &value, 0) == DB_NOTFOUND) { - THROW_STORE_EXCEPTION("Cannot load content. Message not known to store!"); - } - //read header only to begin with - headerSize = value.buffer.getLong(); - BufferValue header(headerSize, preamble_length); - if (messageDb.get(0, &key, &header, 0) == DB_NOTFOUND) { - THROW_STORE_EXCEPTION("Cannot load content. Message not known to store!"); + TxnCtxt txn; + txn.begin(env); + try { + if (messageDb.get(txn.get(), &key, &value, 0) == DB_NOTFOUND) { + txn.abort(); + THROW_STORE_EXCEPTION("Cannot load content. Message not known to store!"); + } + + //read header only to begin with + headerSize = value.buffer.getLong(); + + BufferValue header(headerSize, preamble_length); + if (messageDb.get(txn.get(), &key, &header, 0) == DB_NOTFOUND) { + txn.abort(); + THROW_STORE_EXCEPTION("Cannot load content. Message not known to store!"); + } + ret = recovery.recoverMessage(header.buffer); + txn.commit(); + } catch (const DbException& e) { + txn.abort(); + THROW_STORE_EXCEPTION("Unexpected BDB error in BdbMessageStore::getExternMessage(): " + string(e.what())); } - - return recovery.recoverMessage(header.buffer); + return ret; } int BdbMessageStore::enqueueMessage(TxnCtxt& txn, @@ -963,7 +977,17 @@ u_int64_t BdbMessageStore::getRecordSize(Db& db, Dbt& key) { - return getRecordSize(0, db, key); + u_int64_t ret = 0; + TxnCtxt txn; + txn.begin(env); + try { + ret = getRecordSize(txn.get(), db, key); + txn.commit(); + } catch (const std::exception& e) { + txn.abort(); + throw; + } + return ret; } u_int64_t BdbMessageStore::getRecordSize(DbTxn* txn, Modified: store/trunk/cpp/lib/BdbMessageStore.h =================================================================== --- store/trunk/cpp/lib/BdbMessageStore.h 2008-07-29 20:13:17 UTC (rev 2232) +++ store/trunk/cpp/lib/BdbMessageStore.h 2008-07-30 15:00:18 UTC (rev 2233) @@ -115,6 +115,7 @@ static qpid::sys::Duration defJournalFlushTimeout; qpid::management::Store* mgmtObject; qpid::sys::Mutex jrnlCreateLock; + qpid::sys::Mutex bdbQueueLock; // Parameter validation and calculation static u_int16_t chkJrnlNumFilesParam(const u_int16_t param,

15 years, 10 months

1
0
0 / 0

2024

2023

2022

2021

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011

2010

2009

2008

2007

rhmessaging-commits July 2008