Author: kpvdr Date: 2010-12-02 15:48:07 -0500 (Thu, 02 Dec 2010) New Revision: 4420 Added: store/trunk/cpp/lib/jrnl2/ store/trunk/cpp/lib/jrnl2/Makefile.am store/trunk/cpp/lib/jrnl2/README store/trunk/cpp/lib/jrnl2/aio_callback.hpp store/trunk/cpp/lib/jrnl2/dtok.cpp store/trunk/cpp/lib/jrnl2/dtok.hpp store/trunk/cpp/lib/jrnl2/dtok_state.cpp store/trunk/cpp/lib/jrnl2/dtok_state.hpp store/trunk/cpp/lib/jrnl2/jdir.cpp store/trunk/cpp/lib/jrnl2/jdir.hpp store/trunk/cpp/lib/jrnl2/jerrno.cpp store/trunk/cpp/lib/jrnl2/jerrno.hpp store/trunk/cpp/lib/jrnl2/jexception.cpp store/trunk/cpp/lib/jrnl2/jexception.hpp store/trunk/cpp/lib/jrnl2/jrnl.cpp store/trunk/cpp/lib/jrnl2/jrnl.hpp store/trunk/cpp/lib/jrnl2/jrnl_state.cpp store/trunk/cpp/lib/jrnl2/jrnl_state.hpp store/trunk/cpp/lib/jrnl2/slock.cpp store/trunk/cpp/lib/jrnl2/slock.hpp store/trunk/cpp/lib/jrnl2/smutex.cpp store/trunk/cpp/lib/jrnl2/smutex.hpp store/trunk/cpp/lib/jrnl2/store_params.cpp store/trunk/cpp/lib/jrnl2/store_params.hpp store/trunk/cpp/lib/jrnl2/txn_ctxt.hpp store/trunk/cpp/perf/JournalInstance.cpp store/trunk/cpp/perf/JournalInstance.hpp store/trunk/cpp/perf/JournalParameters.cpp store/trunk/cpp/perf/JournalParameters.hpp store/trunk/cpp/perf/PerformanceTimer.cpp store/trunk/cpp/perf/PerformanceTimer.hpp store/trunk/cpp/perf/StorePerformanceTest.cpp store/trunk/cpp/perf/StorePerformanceTest.hpp store/trunk/cpp/perf/TestParameters.cpp store/trunk/cpp/perf/TestParameters.hpp store/trunk/cpp/perf/m Removed: store/trunk/cpp/perf/perf.cpp store/trunk/cpp/perf/perf.hpp store/trunk/cpp/perf/perf_test.cpp store/trunk/cpp/perf/perf_test.hpp store/trunk/cpp/perf/perf_timer.cpp store/trunk/cpp/perf/perf_timer.hpp store/trunk/cpp/perf/store_params.cpp store/trunk/cpp/perf/store_params.hpp store/trunk/cpp/perf/test_params.cpp store/trunk/cpp/perf/test_params.hpp Modified: store/trunk/cpp/Makefile.am store/trunk/cpp/configure.ac store/trunk/cpp/lib/Makefile.am store/trunk/cpp/lib/jrnl/jcntl.cpp store/trunk/cpp/perf/Makefile.am Log: Update to perf test code and added an experimental null store Modified: store/trunk/cpp/Makefile.am =================================================================== --- store/trunk/cpp/Makefile.am 2010-12-02 15:11:07 UTC (rev 4419) +++ store/trunk/cpp/Makefile.am 2010-12-02 20:48:07 UTC (rev 4420) @@ -28,7 +28,7 @@ sysconf_DATA = etc/rhmd.conf -SUBDIRS = lib tests tools docs +SUBDIRS = lib tests tools perf perf2 docs # Update libtool, if needed. libtool: $(LIBTOOL_DEPS) Modified: store/trunk/cpp/configure.ac =================================================================== --- store/trunk/cpp/configure.ac 2010-12-02 15:11:07 UTC (rev 4419) +++ store/trunk/cpp/configure.ac 2010-12-02 20:48:07 UTC (rev 4420) @@ -217,14 +217,16 @@ AC_CONFIG_FILES([ Makefile + docs/Makefile + lib/Makefile + lib/jrnl2/Makefile + perf/Makefile + perf2/Makefile tests/Makefile tests/cluster/Makefile tests/jrnl/Makefile tests/jrnl/jtt/Makefile - perf/Makefile tools/Makefile - lib/Makefile - docs/Makefile ]) AC_OUTPUT Modified: store/trunk/cpp/lib/Makefile.am =================================================================== --- store/trunk/cpp/lib/Makefile.am 2010-12-02 15:11:07 UTC (rev 4419) +++ store/trunk/cpp/lib/Makefile.am 2010-12-02 20:48:07 UTC (rev 4420) @@ -19,9 +19,10 @@ # # The GNU Lesser General Public License is available in the file COPYING. -AM_CXXFLAGS = $(WARNING_CFLAGS) $(APR_CXXFLAGS) $(QPID_CXXFLAGS) \ - -I$(srcdir)/gen -DRHM_CLEAN -pthread +AM_CXXFLAGS = $(WARNING_CFLAGS) $(APR_CXXFLAGS) $(QPID_CXXFLAGS) -I$(srcdir)/gen -DRHM_CLEAN -pthread +SUBDIRS = jrnl2 + # Definitions for client and daemon plugins PLUGINLDFLAGS=-no-undefined -module -avoid-version dmoduledir=$(libdir)/qpid/daemon Modified: store/trunk/cpp/lib/jrnl/jcntl.cpp =================================================================== --- store/trunk/cpp/lib/jrnl/jcntl.cpp 2010-12-02 15:11:07 UTC (rev 4419) +++ store/trunk/cpp/lib/jrnl/jcntl.cpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -212,27 +212,26 @@ jcntl::enqueue_data_record(const void* const data_buff, const std::size_t tot_data_len, const std::size_t this_data_len, data_tok* dtokp, const bool transient) { + iores r; check_wstatus("enqueue_data_record"); { slock s(_wr_mutex); - iores r; - while (handle_aio_wait(_wmgr.enqueue(data_buff, tot_data_len, this_data_len, dtokp, 0, - 0, transient, false), r, dtokp)) ; - return r; + while (handle_aio_wait(_wmgr.enqueue(data_buff, tot_data_len, this_data_len, dtokp, 0, 0, transient, false), r, + dtokp)) ; } + return r; } iores jcntl::enqueue_extern_data_record(const std::size_t tot_data_len, data_tok* dtokp, const bool transient) { + iores r; check_wstatus("enqueue_extern_data_record"); { slock s(_wr_mutex); - iores r; - while (handle_aio_wait(_wmgr.enqueue(0, tot_data_len, 0, dtokp, 0, 0, transient, - true), r, dtokp)) ; - return r; + while (handle_aio_wait(_wmgr.enqueue(0, tot_data_len, 0, dtokp, 0, 0, transient, true), r, dtokp)) ; } + return r; } iores @@ -240,28 +239,28 @@ const std::size_t this_data_len, data_tok* dtokp, const std::string& xid, const bool transient) { + iores r; check_wstatus("enqueue_tx_data_record"); { slock s(_wr_mutex); - iores r; - while (handle_aio_wait(_wmgr.enqueue(data_buff, tot_data_len, this_data_len, dtokp, - xid.data(), xid.size(), transient, false), r, dtokp)) ; - return r; + while (handle_aio_wait(_wmgr.enqueue(data_buff, tot_data_len, this_data_len, dtokp, xid.data(), xid.size(), + transient, false), r, dtokp)) ; } + return r; } iores jcntl::enqueue_extern_txn_data_record(const std::size_t tot_data_len, data_tok* dtokp, const std::string& xid, const bool transient) { + iores r; check_wstatus("enqueue_extern_txn_data_record"); { slock s(_wr_mutex); - iores r; - while (handle_aio_wait(_wmgr.enqueue(0, tot_data_len, 0, dtokp, xid.data(), xid.size(), - transient, true), r, dtokp)) ; - return r; + while (handle_aio_wait(_wmgr.enqueue(0, tot_data_len, 0, dtokp, xid.data(), xid.size(), transient, true), r, + dtokp)) ; } + return r; } /* TODO @@ -302,56 +301,57 @@ iores jcntl::dequeue_data_record(data_tok* const dtokp, const bool txn_coml_commit) { + iores r; check_wstatus("dequeue_data"); { slock s(_wr_mutex); - iores r; while (handle_aio_wait(_wmgr.dequeue(dtokp, 0, 0, txn_coml_commit), r, dtokp)) ; - return r; } + return r; } iores jcntl::dequeue_txn_data_record(data_tok* const dtokp, const std::string& xid, const bool txn_coml_commit) { + iores r; check_wstatus("dequeue_data"); { slock s(_wr_mutex); - iores r; while (handle_aio_wait(_wmgr.dequeue(dtokp, xid.data(), xid.size(), txn_coml_commit), r, dtokp)) ; - return r; } + return r; } iores jcntl::txn_abort(data_tok* const dtokp, const std::string& xid) { + iores r; check_wstatus("txn_abort"); { slock s(_wr_mutex); - iores r; while (handle_aio_wait(_wmgr.abort(dtokp, xid.data(), xid.size()), r, dtokp)) ; - return r; } + return r; } iores jcntl::txn_commit(data_tok* const dtokp, const std::string& xid) { + iores r; check_wstatus("txn_commit"); { slock s(_wr_mutex); - iores r; while (handle_aio_wait(_wmgr.commit(dtokp, xid.data(), xid.size()), r, dtokp)) ; - return r; } + return r; } bool jcntl::is_txn_synced(const std::string& xid) { slock s(_wr_mutex); - return _wmgr.is_txn_synced(xid); + bool res = _wmgr.is_txn_synced(xid); + return res; } int32_t @@ -360,7 +360,8 @@ stlock t(_wr_mutex); if (!t.locked()) return jerrno::LOCK_TAKEN; - return _wmgr.get_events(pmgr::UNUSED, timeout); + int32_t res = _wmgr.get_events(pmgr::UNUSED, timeout); + return res; } int32_t @@ -497,8 +498,15 @@ void jcntl::aio_cmpl_wait() { - while (_wmgr.get_aio_evt_rem()) + //while (_wmgr.get_aio_evt_rem()) + while (true) { + u_int32_t aer; + { + slock s(_wr_mutex); + aer = _wmgr.get_aio_evt_rem(); + } + if (aer == 0) break; // no events left if (get_wr_events(&_aio_cmpl_timeout) == jerrno::AIO_TIMEOUT) throw jexception(jerrno::JERR_JCNTL_AIOCMPLWAIT, "jcntl", "aio_cmpl_wait"); } Added: store/trunk/cpp/lib/jrnl2/Makefile.am =================================================================== --- store/trunk/cpp/lib/jrnl2/Makefile.am (rev 0) +++ store/trunk/cpp/lib/jrnl2/Makefile.am 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,49 @@ +# Copyright (c) 2007, 2008, 2009 Red Hat, Inc. +# +# This file is part of the Qpid async store library msgstore.so. +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 +# USA +# +# The GNU Lesser General Public License is available in the file COPYING. + +AM_CXXFLAGS = $(WARNING_CFLAGS) -pthread + +lib_LTLIBRARIES = libasyncjrnl2.la + +libasyncjrnl2_la_SOURCES = \ + dtok.cpp \ + dtok_state.cpp \ + jdir.cpp \ + jerrno.cpp \ + jexception.cpp \ + jrnl.cpp \ + jrnl_state.cpp \ + slock.cpp \ + smutex.cpp \ + store_params.cpp \ + aio_callback.hpp \ + dtok.hpp \ + dtok_state.hpp \ + jdir.hpp \ + jerrno.hpp \ + jexception.hpp \ + jrnl.hpp \ + jrnl_state.hpp \ + slock.hpp \ + smutex.hpp \ + store_params.hpp \ + txn_ctxt.hpp + Added: store/trunk/cpp/lib/jrnl2/README =================================================================== --- store/trunk/cpp/lib/jrnl2/README (rev 0) +++ store/trunk/cpp/lib/jrnl2/README 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,3 @@ +This directory contains experimental code. It is not (currently) +a part of the main store, and can be safely ignored for all +normal builds. Added: store/trunk/cpp/lib/jrnl2/aio_callback.hpp =================================================================== --- store/trunk/cpp/lib/jrnl2/aio_callback.hpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/aio_callback.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,56 @@ +/** + * \file aio_callback.hpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#ifndef mrg_journal2_aio_callback_hpp +#define mrg_journal2_aio_callback_hpp + +#include "dtok.hpp" +#include <sys/types.h> // u_int16_t, u_int32_t +#include <vector> + +namespace mrg +{ +namespace journal2 +{ + + class aio_callback + { + public: + virtual void wr_aio_cb(std::vector<dtok*>& dtokl) = 0; + virtual void rd_aio_cb(std::vector<u_int16_t>& pil) = 0; + }; + +} // namespace journal2 +} // namespace mrg + + +#endif // mrg_journal2_aio_callback_hpp + Added: store/trunk/cpp/lib/jrnl2/dtok.cpp =================================================================== --- store/trunk/cpp/lib/jrnl2/dtok.cpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/dtok.cpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,69 @@ +/** + * \file dtok.cpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#include "dtok.hpp" + +namespace mrg +{ +namespace journal2 +{ +// static +rec_id rid_counter::_rid; + +// static +smutex rid_counter::_rid_mutex; + +dtok::dtok() : _dtok_state(), _rid(rid_counter::get_next_rid()), _external_rid(false) {} + +dtok::dtok(const rec_id rid) : _dtok_state(), _rid(rid), _external_rid(true) {} + +void +dtok::set_rid(const rec_id rid) +{ + _rid = rid; + _external_rid = true; +} + +void +dtok::set_drid(const rec_id drid) +{ + _drid = drid; +} + +std::string +dtok::status_str() +{ + return "status string"; +} + +} // namespace journal2 +} // namespace mrg + Added: store/trunk/cpp/lib/jrnl2/dtok.hpp =================================================================== --- store/trunk/cpp/lib/jrnl2/dtok.hpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/dtok.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,92 @@ +/** + * \file dtok.hpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#ifndef mrg_journal2_dtok_hpp +#define mrg_journal2_dtok_hpp + +#include <string> +#include <sys/types.h> // u_int16_t, u_int32_t + +#include "dtok_state.hpp" +#include "slock.hpp" +#include "smutex.hpp" + +namespace mrg +{ +namespace journal2 +{ + + typedef u_int64_t rec_id; + + class rid_counter + { + protected: + static rec_id _rid; + static smutex _rid_mutex; + public: + inline static rec_id get_next_rid() { slock l(_rid_mutex); return ++_rid; } + }; + + class dtok + { + protected: + dtok_state _dtok_state; + bool _transient; + bool _external; + std::string _external_location; + rec_id _rid; + bool _external_rid; + rec_id _drid; + + public: + dtok(); + dtok(const rec_id rid); + + inline dtok_state& get_dtok_state() { return _dtok_state; } + inline bool is_transient() const { return _transient; } + inline bool is_external() const { return _external; } + inline std::string& get_external_location() { return _external_location; } + void set_rid(const rec_id rid); + inline rec_id rid() const { return _rid; } + inline bool is_external_rid() { return _external_rid; } + void set_drid(const rec_id drid); + inline rec_id drid() const { return _drid; } + + // debug aids + std::string status_str(); + }; + +} // namespace journal2 +} // namespace mrg + + +#endif // mrg_journal2_dtok_hpp + Added: store/trunk/cpp/lib/jrnl2/dtok_state.cpp =================================================================== --- store/trunk/cpp/lib/jrnl2/dtok_state.cpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/dtok_state.cpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,237 @@ +/** + * \file dtok_state.cpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#include "dtok_state.hpp" + +#include "jexception.hpp" + +namespace mrg +{ +namespace journal2 +{ + +dtok_state::dtok_state() : + _op_state(OP_NONE), + _txn_state(TX_NONE), + _buff_state(IO_NONE), + _iosubm_state(IO_NONE), + _iocmpl_state(IO_NONE) +{} + +dtok_state::dtok_state(const dtok_state& s) : + _op_state(s._op_state), + _txn_state(s._txn_state), + _buff_state(s._buff_state), + _iosubm_state(s._iosubm_state), + _iocmpl_state(s._iocmpl_state) +{} + +dtok_state::dtok_state(const opstate_t o, + const txnstate_t t, + const iostate_t b, + const iostate_t s, + const iostate_t c) : + _op_state(o), + _txn_state(t), + _buff_state(b), + _iosubm_state(s), + _iocmpl_state(c) +{} + +void +dtok_state::reset() +{ + _op_state = OP_NONE; + _txn_state = TX_NONE; + _buff_state = IO_NONE; + _iosubm_state = IO_NONE; + _iocmpl_state = IO_NONE; +} + +void +dtok_state::set_enqueue() +{ + if (_op_state != OP_NONE) + THROW_STATE_EXCEPTION(jerrno::JERR_BADDTOKOPSTATE, + get_state_str(OP_ENQUEUE), + get_state_str(_op_state), + "dtok_state", + "set_enqueue"); + _op_state = OP_ENQUEUE; +} + +void +dtok_state::set_dequeue() +{ + if (_op_state != OP_ENQUEUE) + THROW_STATE_EXCEPTION(jerrno::JERR_BADDTOKOPSTATE, + get_state_str(OP_DEQUEUE), + get_state_str(_op_state), + "dtok_state", + "set_dequeue"); + _op_state = OP_DEQUEUE; +} + +void +dtok_state::set_txn_commit() +{ + if (_txn_state != TX_NONE) + THROW_STATE_EXCEPTION(jerrno::JERR_BADDTOKTXNSTATE, + get_state_str(TX_COMMIT), + get_state_str(_txn_state), + "dtok_state", + "set_txn_commit"); + _txn_state = TX_COMMIT; +} + +void +dtok_state::set_txn_abort() +{ + if (_txn_state != TX_NONE) + THROW_STATE_EXCEPTION(jerrno::JERR_BADDTOKTXNSTATE, + get_state_str(TX_ABORT), + get_state_str(_txn_state), + "dtok_state", + "set_txn_abort"); + _txn_state = TX_ABORT; +} + +void +dtok_state::set_buff_part() +{ + if (_buff_state != IO_NONE) + THROW_STATE_EXCEPTION(jerrno::JERR_BADDTOKIOSTATE, + get_state_str(IO_PART), + get_state_str(_buff_state), + "dtok_state", + "set_buff_part"); + _buff_state = IO_PART; +} + +void +dtok_state::set_buff_compl() +{ + _buff_state = IO_COMPLETE; +} + +void +dtok_state::set_iosubm_part() +{ + if (_iosubm_state != IO_NONE) + THROW_STATE_EXCEPTION(jerrno::JERR_BADDTOKIOSTATE, + get_state_str(IO_PART), + get_state_str(_iosubm_state), + "dtok_state", + "set_iosubm_part"); + _iosubm_state = IO_PART; +} + +void +dtok_state::set_iosubm_compl() +{ + _iosubm_state = IO_COMPLETE; +} + +void +dtok_state::set_iocompl_part() +{ + if (_iocmpl_state != IO_NONE) + THROW_STATE_EXCEPTION(jerrno::JERR_BADDTOKIOSTATE, + get_state_str(IO_PART), + get_state_str(_iocmpl_state), + "dtok_state", + "set_iocompl_part"); + _iocmpl_state = IO_PART; +} + +void +dtok_state::set_iocompl_compl() +{ + _iocmpl_state = IO_COMPLETE; +} + +//static +std::string +dtok_state::get_state_str(opstate_t s) +{ + switch (s) { + case OP_NONE: + return "OP_NONE"; + case OP_ENQUEUE: + return "OP_ENQUEUE"; + case OP_DEQUEUE: + return "OP_DEQUEUE"; + default: + std::ostringstream oss; + oss << "<unknown state (" << s << ")>"; + return oss.str(); + } +} + +//static +std::string +dtok_state::get_state_str(txnstate_t s) +{ + switch (s) { + case TX_NONE: + return "TX_NONE"; + case TX_COMMIT: + return "TX_COMMIT"; + case TX_ABORT: + return "TX_ABORT"; + default: + std::ostringstream oss; + oss << "<unknown state (" << s << ")>"; + return oss.str(); + } +} + +//static +std::string +dtok_state::get_state_str(iostate_t s) +{ + switch (s) { + case IO_NONE: + return "IO_NONE"; + case IO_PART: + return "IO_PART"; + case IO_COMPLETE: + return "IO_COMPLETE"; + default: + std::ostringstream oss; + oss << "<unknown state (" << s << ")>"; + return oss.str(); + } +} + +} // namespace journal2 +} // namespace mrg + Added: store/trunk/cpp/lib/jrnl2/dtok_state.hpp =================================================================== --- store/trunk/cpp/lib/jrnl2/dtok_state.hpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/dtok_state.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,124 @@ +/** + * \file dtok_state.hpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#ifndef mrg_journal2_dtok_state_hpp +#define mrg_journal2_dtok_state_hpp + +#include <string> + +namespace mrg +{ +namespace journal2 +{ + + typedef enum + { + OP_NONE = 0, + OP_ENQUEUE, + OP_DEQUEUE + } opstate_t; + + typedef enum + { + TX_NONE = 0, + TX_COMMIT, + TX_ABORT + } txnstate_t; + + typedef enum + { + IO_NONE = 0, + IO_PART, + IO_COMPLETE + } iostate_t; + + class dtok_state + { + protected: + opstate_t _op_state; + txnstate_t _txn_state; + iostate_t _buff_state; + iostate_t _iosubm_state; + iostate_t _iocmpl_state; + public: + dtok_state(); + dtok_state(const dtok_state& s); + dtok_state(const opstate_t o, + const txnstate_t t, + const iostate_t b, + const iostate_t s, + const iostate_t c); + + // Raw state get/set functions + inline opstate_t get_op_state() const { return _op_state; } + inline void set_op_state(const opstate_t s) { _op_state = s; } + inline txnstate_t get_txn_state() const { return _txn_state; } + inline void set_txn_state(const txnstate_t t) { _txn_state = t; } + inline iostate_t get_buff_state() const { return _buff_state; } + inline void set_buff_state(const iostate_t b) { _buff_state = b; } + inline iostate_t get_iosubm_state() const { return _iosubm_state; } + inline void set_iosubm_state(const iostate_t s) { _iosubm_state = s; } + inline iostate_t get_iocompl_state() const { return _iocmpl_state; } + inline void set_iocompl_state(const iostate_t c) { _iocmpl_state = c; } + + // State change functions + void reset(); + void set_enqueue(); + void set_dequeue(); + void set_txn_commit(); + void set_txn_abort(); + void set_buff_part(); + void set_buff_compl(); + void set_iosubm_part(); + void set_iosubm_compl(); + void set_iocompl_part(); + void set_iocompl_compl(); + + // State query functions + inline bool is_enqueueable() const { return _op_state == OP_NONE; } + inline bool is_dequeueable() const { return _op_state == OP_ENQUEUE; } + inline bool is_txn_open() const { return _txn_state == TX_NONE; } + inline bool is_buff_write_compl() const { return _buff_state == IO_COMPLETE; } + inline bool is_iosubm_compl() const { return _iosubm_state == IO_COMPLETE; } + inline bool is_iocompl_compl() const { return _iocmpl_state == IO_COMPLETE; } + + // State-to-string function(s) + static std::string get_state_str(opstate_t s); + static std::string get_state_str(txnstate_t s); + static std::string get_state_str(iostate_t s); + }; + +} // namespace journal2 +} // namespace mrg + + +#endif // mrg_journal2_dtok_state_hpp + Added: store/trunk/cpp/lib/jrnl2/jdir.cpp =================================================================== --- store/trunk/cpp/lib/jrnl2/jdir.cpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/jdir.cpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,84 @@ +/** + * \file jdir.cpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#include "jdir.hpp" + +namespace mrg +{ +namespace journal2 +{ + +// static +bool +jdir::exists(const std::string& name) +{ + return exists(name.c_str()); +} + +// static +bool +jdir::exists(const char* /*name*/) +{ + // TODO - add impl here + return false; +} + +// static +void +jdir::create_dir(const std::string& name) +{ + return create_dir(name.c_str()); +} + +// static +void +jdir::create_dir(const char* /*name*/) +{ + // TODO - add impl here +} + +// static +void +jdir::delete_dir(const std::string& name) +{ + return delete_dir(name.c_str()); +} + +// static +void +jdir::delete_dir(const char* /*name*/) +{ + // TODO - add impl here +} + + +} // namespace journal2 +} // namespace mrg Added: store/trunk/cpp/lib/jrnl2/jdir.hpp =================================================================== --- store/trunk/cpp/lib/jrnl2/jdir.hpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/jdir.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,59 @@ +/** + * \file jdir.hpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#ifndef mrg_journal2_jdir_hpp +#define mrg_journal2_jdir_hpp + +#include <string> + +namespace mrg +{ +namespace journal2 +{ + + class jdir + { + public: + static bool exists(const std::string& name); + static bool exists(const char* name); + static void create_dir(const std::string& name); + static void create_dir(const char* name); + static void delete_dir(const std::string& name); + static void delete_dir(const char* name); + }; + + +} // namespace journal2 +} // namespace mrg + + +#endif // mrg_journal2_jdir_hpp + Added: store/trunk/cpp/lib/jrnl2/jerrno.cpp =================================================================== --- store/trunk/cpp/lib/jrnl2/jerrno.cpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/jerrno.cpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,74 @@ +/** + * \file jerrno.cpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#include "jerrno.hpp" + +namespace mrg +{ +namespace journal2 +{ + +std::map<u_int32_t, const char*> jerrno::_err_map; +std::map<u_int32_t, const char*>::iterator jerrno::_err_map_itr; +bool jerrno::_initialized = jerrno::__init(); + +// generic errors +const u_int32_t jerrno::JERR_PTHREAD = 0x0001; + +// illegal states +const u_int32_t jerrno::JERR_BADJRNLSTATE = 0x0101; +const u_int32_t jerrno::JERR_BADDTOKOPSTATE = 0x0102; +const u_int32_t jerrno::JERR_BADDTOKTXNSTATE = 0x0103; +const u_int32_t jerrno::JERR_BADDTOKIOSTATE = 0x0104; + +bool +jerrno::__init() +{ + _err_map[JERR_PTHREAD] = "JERR_PTHREAD: pthread operation failure"; + _err_map[JERR_BADJRNLSTATE] = "JERR_BADJRNLSTATE: Illegal journal state"; + _err_map[JERR_BADDTOKOPSTATE] = "JERR_BADDTOKOPSTATE: Illegal data token op state"; + _err_map[JERR_BADDTOKTXNSTATE] = "JERR_BADDTOKTXNSTATE: Illegal data token txn state"; + _err_map[JERR_BADDTOKIOSTATE] = "JERR_BADDTOKIOSTATE: Illegal data token io state"; + + return true; +} + +const char* +jerrno::err_msg(const u_int32_t err_no) throw () +{ + _err_map_itr = _err_map.find(err_no); + if (_err_map_itr == _err_map.end()) + return "<Unknown error code>"; + return _err_map_itr->second; +} + +} // namespace journal2 +} // namespace mrg Added: store/trunk/cpp/lib/jrnl2/jerrno.hpp =================================================================== --- store/trunk/cpp/lib/jrnl2/jerrno.hpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/jerrno.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,67 @@ +/** + * \file jerrno.hpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#ifndef mrg_journal2_jerrno_hpp +#define mrg_journal2_jerrno_hpp + +#include <map> +#include <sys/types.h> // u_int16_t, u_int32_t + +namespace mrg +{ +namespace journal2 +{ + + class jerrno + { + protected: + static std::map<u_int32_t, const char*> _err_map; ///< Map of error messages + static std::map<u_int32_t, const char*>::iterator _err_map_itr; ///< Iterator + static bool _initialized; ///< Dummy flag, used to initialize map. + static bool __init(); ///< Static fn for initializing static data + public: + // generic errors + static const u_int32_t JERR_PTHREAD; ///< pthread operation failure + + // illegal states + static const u_int32_t JERR_BADJRNLSTATE; ///< Illegal journal state + static const u_int32_t JERR_BADDTOKOPSTATE; ///< Illegal data token op state + static const u_int32_t JERR_BADDTOKTXNSTATE; ///< Illegal data token txn state + static const u_int32_t JERR_BADDTOKIOSTATE; ///< Illegal data token io state + + + static const char* err_msg(const u_int32_t err_no) throw (); + }; + +} // namespace journal2 +} // namespace mrg + +#endif // mrg_journal2_jerrno_hpp Added: store/trunk/cpp/lib/jrnl2/jexception.cpp =================================================================== --- store/trunk/cpp/lib/jrnl2/jexception.cpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/jexception.cpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,187 @@ +/** + * \file jexception.cpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#include "jexception.hpp" + +#include <iomanip> +#include "jerrno.hpp" + +namespace mrg +{ +namespace journal2 +{ + +jexception::jexception() throw (): + std::exception(), + _err_code(0) +{ + format(); +} + +jexception::jexception(const u_int32_t err_code) throw (): + std::exception(), + _err_code(err_code) +{ + format(); +} + +jexception::jexception(const char* additional_info) throw (): + std::exception(), + _err_code(0), + _additional_info(additional_info) +{ + format(); +} + +jexception::jexception(const std::string& additional_info) throw (): + std::exception(), + _err_code(0), + _additional_info(additional_info) +{ + format(); +} + +jexception::jexception(const u_int32_t err_code, + const char* additional_info) throw (): + std::exception(), + _err_code(err_code), + _additional_info(additional_info) +{ + format(); +} + +jexception::jexception(const u_int32_t err_code, + const std::string& additional_info) throw (): + std::exception(), + _err_code(err_code), + _additional_info(additional_info) +{ + format(); +} + +jexception::jexception(const u_int32_t err_code, + const char* throwing_class, + const char* throwing_fn) throw (): + std::exception(), + _err_code(err_code), + _throwing_class(throwing_class), + _throwing_fn(throwing_fn) +{ + format(); +} + +jexception::jexception(const u_int32_t err_code, + const std::string& throwing_class, + const std::string& throwing_fn) throw (): + std::exception(), + _err_code(err_code), + _throwing_class(throwing_class), + _throwing_fn(throwing_fn) +{ + format(); +} + +jexception::jexception(const u_int32_t err_code, + const char* additional_info, + const char* throwing_class, + const char* throwing_fn) throw (): + std::exception(), + _err_code(err_code), + _additional_info(additional_info), + _throwing_class(throwing_class), + _throwing_fn(throwing_fn) +{ + format(); +} + +jexception::jexception(const u_int32_t err_code, + const std::string& additional_info, + const std::string& throwing_class, + const std::string& throwing_fn) throw (): + std::exception(), + _err_code(err_code), + _additional_info(additional_info), + _throwing_class(throwing_class), + _throwing_fn(throwing_fn) +{ + format(); +} + +jexception::~jexception() throw () +{} + +void +jexception::format() +{ + const bool ai = !_additional_info.empty(); + const bool tc = !_throwing_class.empty(); + const bool tf = !_throwing_fn.empty(); + std::ostringstream oss; + oss << "jexception 0x" << std::hex << std::setfill('0') << std::setw(4) << _err_code << " "; + if (tc) + { + oss << _throwing_class; + if (tf) + oss << "::"; + else + oss << " "; + } + if (tf) + oss << _throwing_fn << "() "; + if (tc || tf) + oss << "threw " << jerrno::err_msg(_err_code); + if (ai) + oss << " (" << _additional_info << ")"; + _what.assign(oss.str()); +} + +const char* +jexception::what() const throw () +{ + return _what.c_str(); +} + +std::ostream& +operator<<(std::ostream& os, const jexception& je) +{ + os << je.what(); + return os; +} + +std::ostream& +operator<<(std::ostream& os, const jexception* jePtr) +{ + os << jePtr->what(); + return os; +} + +} // namespace journal2 +} // namespace mrg Added: store/trunk/cpp/lib/jrnl2/jexception.hpp =================================================================== --- store/trunk/cpp/lib/jrnl2/jexception.hpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/jexception.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,118 @@ +/** + * \file jexception.hpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#ifndef mrg_journal2_jexception_hpp +#define mrg_journal2_jexception_hpp + +#include <cstring> +#include <sstream> +#include <string> +#include <sys/types.h> // u_int16_t, u_int32_t + +#include "jerrno.hpp" + +// Macro definitions + +#define FORMAT_SYSERR(errno) " errno=" << errno << " (" << std::strerror(errno) << ")" + +#define PTHREAD_CHK(err, pfn, cls, fn) if(err != 0) { \ + std::ostringstream oss; \ + oss << pfn << " failed: " << FORMAT_SYSERR(err); \ + throw jexception(jerrno::JERR_PTHREAD, oss.str(), cls, fn); \ + } + +#define THROW_STATE_EXCEPTION(jerrno, target_st, curr_st, cls, fn) { \ + std::ostringstream oss; \ + oss << cls << "::" << fn << "() in state " << curr_st << " cannot be moved to state " << target_st << "."; \ + throw jexception(jerrno, oss.str(), cls, fn); \ + } + +namespace mrg +{ +namespace journal2 +{ + + class jexception : public std::exception + { + protected: + u_int32_t _err_code; + std::string _additional_info; + std::string _throwing_class; + std::string _throwing_fn; + std::string _what; + void format(); + + public: + jexception() throw (); + + jexception(const u_int32_t err_code) throw (); + + jexception(const char* additional_info) throw (); + jexception(const std::string& additional_info) throw (); + + jexception(const u_int32_t err_code, + const char* additional_info) throw (); + jexception(const u_int32_t err_code, + const std::string& additional_info) throw (); + + jexception(const u_int32_t err_code, + const char* throwing_class, + const char* throwing_fn) throw (); + jexception(const u_int32_t err_code, + const std::string& throwing_class, + const std::string& throwing_fn) + throw (); + + jexception(const u_int32_t err_code, + const char* additional_info, + const char* throwing_class, + const char* throwing_fn) throw (); + jexception(const u_int32_t err_code, + const std::string& additional_info, + const std::string& throwing_class, + const std::string& throwing_fn) throw (); + + virtual ~jexception() throw (); + virtual const char* what() const throw (); // override std::exception::what() + + inline u_int32_t err_code() const throw () { return _err_code; } + inline const std::string additional_info() const throw () { return _additional_info; } + inline const std::string throwing_class() const throw () { return _throwing_class; } + inline const std::string throwing_fn() const throw () { return _throwing_fn; } + + friend std::ostream& operator<<(std::ostream& os, const jexception& je); + friend std::ostream& operator<<(std::ostream& os, const jexception* jePtr); + }; // class jexception + +} // namespace journal2 +} // namespace mrg + +#endif // mrg_journal2_jexception_hpp Added: store/trunk/cpp/lib/jrnl2/jrnl.cpp =================================================================== --- store/trunk/cpp/lib/jrnl2/jrnl.cpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/jrnl.cpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,184 @@ +/** + * \file jrnl.cpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#include "jrnl.hpp" + +// --- temp code --- +#include <iostream> +#include <vector> +// --- end temp code --- + +namespace mrg +{ +namespace journal2 +{ + +std::string +iores_str(const iores /*res*/) +{ + // TODO - provide implementation + return ".[iores_str]."; +} + +// static +u_int32_t jrnl::list_thresh = 50; + +jrnl::jrnl(const std::string& jid, + const std::string& jdir, + const std::string& base_filename) : + _jid(jid), + _jdir(jdir), + _base_filename(base_filename), + _store_params_ptr(0), + _cbp(0) + // --- temp code --- + , cb_dtok_list_switch(false) + // --- end temp code --- +{} + +void +jrnl::initialize(const store_params* sp, aio_callback* const cbp) +{ + _store_params_ptr = sp; + _cbp = cbp; +} + +iores +jrnl::enqueue(const void* const /*msg_ptr*/, const std::size_t /*msg_size*/, dtok* const dtokp) +{ + dtokp->get_dtok_state().set_enqueue(); + // --- temp code --- + bool flush_flag; + { + slock l(wr_dtok_list_lock); + wr_dtok_list.push_back(dtokp); + flush_flag = wr_dtok_list.size() >= list_thresh; + if (flush_flag) flush_nl(false); + } + if (flush_flag) get_wr_events(0); + // --- end temp code --- + return 0; +} + +iores +jrnl::dequeue(dtok* const dtokp) +{ + dtokp->get_dtok_state().set_dequeue(); + dtokp->set_drid(dtokp->rid()); + // --- temp code --- + bool flush_flag; + { + slock l(wr_dtok_list_lock); + wr_dtok_list.push_back(dtokp); + flush_flag = wr_dtok_list.size() >= list_thresh; + if (flush_flag) flush_nl(false); + } + if (flush_flag) get_wr_events(0); + // --- end temp code --- + return 0; +} + +iores +jrnl::commit() +{ + // TODO + return 0; +} + +iores +jrnl::abort() +{ + // TODO + return 0; +} + +u_int32_t +jrnl::get_wr_aio_evt_rem() const +{ + while (true) { + stlock l1(cb_dtok_list_lock); + stlock l2(wr_dtok_list_lock); + if (l1.locked() && l2.locked()) { + return cb_dtok_list[0].size() + cb_dtok_list[1].size(); + } else { + //::usleep(10); + } + }; +} + +void +jrnl::flush(const bool block_till_aio_cmpl) +{ + // --- temp code --- + stlock l(wr_dtok_list_lock); + if (l.locked()) { + flush_nl(block_till_aio_cmpl); + } + // --- end temp code --- +} + +// protected +void +jrnl::flush_nl(const bool /*block_till_aio_cmpl*/) +{ + // --- temp code --- + int i = cb_dtok_list_switch ? 1 : 0; + while (wr_dtok_list.size()) { + cb_dtok_list[i].push_back(wr_dtok_list.back()); + wr_dtok_list.pop_back(); + } + // --- end temp code --- +} + +void +jrnl::get_wr_events(timespec* const /*timeout*/) +{ + // --- temp code --- + stlock l1(cb_dtok_list_lock); + if (l1.locked()) { + int i = cb_dtok_list_switch ? 0 : 1; + if (cb_dtok_list[i].size() && _cbp) { + _cbp->wr_aio_cb(cb_dtok_list[i]); + } + cb_dtok_list[i].clear(); + + // take both locks before allowing switch to change + { + slock l2(wr_dtok_list_lock); + cb_dtok_list_switch = !cb_dtok_list_switch; + } + } + // --- end temp code --- +} + +} // namespace journal2 +} // namespace mrg + Added: store/trunk/cpp/lib/jrnl2/jrnl.hpp =================================================================== --- store/trunk/cpp/lib/jrnl2/jrnl.hpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/jrnl.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,112 @@ +/** + * \file jrnl.hpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#ifndef mrg_journal2_jnrl_hpp +#define mrg_journal2_jnrl_hpp + +#include <string> +#include <sys/types.h> // u_int64_t, u_int32_t, etc. +#include <time.h> // timespec + +#include "aio_callback.hpp" +#include "dtok.hpp" +#include "jrnl_state.hpp" +#include "store_params.hpp" + +// --- temp code --- +#include "slock.hpp" +#include "smutex.hpp" +#include <vector> +// --- end temp code --- + +namespace mrg +{ +namespace journal2 +{ + + // TODO - decide if this is the right place to expose these codes and flags + typedef u_int64_t iores; // TODO - this needs to be expressed as flags + const iores RHM_IORES_ENQCAPTHRESH = 0x1; + const iores RHM_IORES_BUSY = 0x2; + std::string iores_str(const iores /*res*/); + + class jrnl + { + protected: + std::string _jid; + std::string _jdir; + std::string _base_filename; + jrnl_state _jrnl_state; + const store_params* _store_params_ptr; + aio_callback* _cbp; + + // --- temp code --- + static u_int32_t list_thresh; + std::vector<dtok*> wr_dtok_list; + std::vector<dtok*> cb_dtok_list[2]; + bool cb_dtok_list_switch; + smutex wr_dtok_list_lock; + smutex cb_dtok_list_lock; + // --- end temp code --- + + void flush_nl(const bool block_till_aio_cmpl); + public: + jrnl(const std::string& jid, const std::string& jdir, const std::string& base_filename); + + // get functions + inline std::string id() { return _jid; } + inline std::string dir() { return _jdir; } + inline std::string base_filename() { return _base_filename; } + inline const jrnl_state& get_jrnl_state() { return _jrnl_state; } + inline const store_params* get_store_params() const { return _store_params_ptr; } + + // msg ops + void initialize(const store_params* sp, aio_callback* const cbp); + iores enqueue(const void* const msg_ptr, const std::size_t msg_size, dtok* const dtokp); + iores dequeue(dtok* const dtokp); + iores commit(); + iores abort(); + + // aio ops and status + // --- temp code --- + u_int32_t get_wr_aio_evt_rem() const; + // --- end of temp code --- + void flush(const bool block_till_aio_cmpl); + void get_wr_events(timespec* const timeout); + }; + + +} // namespace journal2 +} // namespace mrg + + +#endif // mrg_journal2_jnrl_hpp + Added: store/trunk/cpp/lib/jrnl2/jrnl_state.cpp =================================================================== --- store/trunk/cpp/lib/jrnl2/jrnl_state.cpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/jrnl_state.cpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,180 @@ +/** + * \file jrnl_state.cpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#include "jrnl_state.hpp" + +#include "jexception.hpp" + +namespace mrg +{ +namespace journal2 +{ + +jrnl_state::jrnl_state() : _jrnl_state(JS_NONE) +{} + +jrnl_state::jrnl_state(const jrnl_state& s) : _jrnl_state(s._jrnl_state) +{} + +jrnl_state::jrnl_state(const jrnl_state_t s) : _jrnl_state(s) +{} + +const jrnl_state_t& +jrnl_state::get() const +{ + return _jrnl_state; +} + +void +jrnl_state::set(const jrnl_state_t s) +{ + _jrnl_state = s; +} + +void +jrnl_state::reset() +{ + _jrnl_state = JS_NONE; +} + +void +jrnl_state::initialize() +{ + if (_jrnl_state != JS_NONE) + THROW_STATE_EXCEPTION(jerrno::JERR_BADJRNLSTATE, + get_state_str(JS_INITIALIZING), + get_state_str(_jrnl_state), + "jrnl_state", + "initialize"); + _jrnl_state = JS_INITIALIZING; +} + +void +jrnl_state::initialize_compl() +{ + if (_jrnl_state != JS_INITIALIZING) + THROW_STATE_EXCEPTION(jerrno::JERR_BADJRNLSTATE, + get_state_str(JS_RUNNING), + get_state_str(_jrnl_state), + "jrnl_state", + "initialize_compl"); + _jrnl_state = JS_RUNNING; +} + +void +jrnl_state::recover_phase_1() +{ + if (_jrnl_state != JS_NONE) + THROW_STATE_EXCEPTION(jerrno::JERR_BADJRNLSTATE, + get_state_str(JS_RECOVERING_PHASE_1), + get_state_str(_jrnl_state), + "jrnl_state", + "recover_phase_1"); + _jrnl_state = JS_RECOVERING_PHASE_1; +} + +void +jrnl_state::recover_phase_2() +{ + if (_jrnl_state != JS_RECOVERING_PHASE_1) + THROW_STATE_EXCEPTION(jerrno::JERR_BADJRNLSTATE, + get_state_str(JS_RECOVERING_PHASE_2), + get_state_str(_jrnl_state), + "jrnl_state", + "recover_phase_2"); + _jrnl_state = JS_RECOVERING_PHASE_2; +} + +void +jrnl_state::recover_compl() +{ + if (_jrnl_state != JS_RECOVERING_PHASE_2) + THROW_STATE_EXCEPTION(jerrno::JERR_BADJRNLSTATE, + get_state_str(JS_RUNNING), + get_state_str(_jrnl_state), + "jrnl_state", + "recover_compl"); + _jrnl_state = JS_RUNNING; +} + +void +jrnl_state::stop() +{ + if (_jrnl_state != JS_RUNNING) + THROW_STATE_EXCEPTION(jerrno::JERR_BADJRNLSTATE, + get_state_str(JS_STOPPING), + get_state_str(_jrnl_state), + "jrnl_state", + "stop"); + _jrnl_state = JS_STOPPING; +} + +void +jrnl_state::stop_compl() +{ + if (_jrnl_state != JS_STOPPING) + THROW_STATE_EXCEPTION(jerrno::JERR_BADJRNLSTATE, + get_state_str(JS_STOPPED), + get_state_str(_jrnl_state), + "jrnl_state", + "stop_compl"); + _jrnl_state = JS_STOPPED; +} + +//static +std::string +jrnl_state::get_state_str(jrnl_state_t s) +{ + switch (s) { + case JS_NONE: + return "JS_NONE"; + case JS_RECOVERING_PHASE_1: + return "JS_RECOVERING_PHASE_1"; + case JS_RECOVERING_PHASE_2: + return "JS_RECOVERING_PHASE_2"; + case JS_INITIALIZING: + return "JS_INITIALIZING"; + case JS_RUNNING: + return "JS_RUNNING"; + case JS_STOPPING: + return "JS_STOPPING"; + case JS_STOPPED: + return "JS_STOPPED"; + default: + std::ostringstream oss; + oss << "<unknown state (" << "s" << ")>"; + return oss.str(); + } +} + +} // namespace journal2 +} // namespace mrg + Added: store/trunk/cpp/lib/jrnl2/jrnl_state.hpp =================================================================== --- store/trunk/cpp/lib/jrnl2/jrnl_state.hpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/jrnl_state.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,104 @@ +/** + * \file jrnl_state.hpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#ifndef mrg_journal2_jrnl_state_hpp +#define mrg_journal2_jrnl_state_hpp + +#include <string> + +namespace mrg +{ +namespace journal2 +{ + + /** + * Journal state machine + * + * JS_NONE * + * / \ + * / v + * | o JS_RECOVERING_PHASE_1 + * v | + * JS_INITIALIZING o v + * | o JS_RECOVERING_PHASE_2 + * \ / + * v v + * JS_RUNNING o + * | + * v + * JS_STOPPING o + * | + * v + * JS_STOPPED * + */ + typedef enum + { + JS_NONE = 0, + JS_RECOVERING_PHASE_1, + JS_RECOVERING_PHASE_2, + JS_INITIALIZING, + JS_RUNNING, + JS_STOPPING, + JS_STOPPED + } jrnl_state_t; + + class jrnl_state + { + protected: + jrnl_state_t _jrnl_state; + public: + jrnl_state(); + jrnl_state(const jrnl_state& s); + jrnl_state(const jrnl_state_t s); + + // Raw state get/set functions + const jrnl_state_t& get() const; + void set(const jrnl_state_t s); + + // State change functions + void reset(); + void initialize(); + void initialize_compl(); + void recover_phase_1(); + void recover_phase_2(); + void recover_compl(); + void stop(); + void stop_compl(); + + static std::string get_state_str(jrnl_state_t s); + }; + +} // namespace journal2 +} // namespace mrg + + +#endif // mrg_journal2_jrnl_state_hpp + Added: store/trunk/cpp/lib/jrnl2/slock.cpp =================================================================== --- store/trunk/cpp/lib/jrnl2/slock.cpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/slock.cpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,32 @@ +/** + * \file slock.cpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#include "slock.hpp" Added: store/trunk/cpp/lib/jrnl2/slock.hpp =================================================================== --- store/trunk/cpp/lib/jrnl2/slock.hpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/slock.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,93 @@ +/** + * \file slock.hpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#ifndef mrg_journal2_slock_hpp +#define mrg_journal2_slock_hpp + +#include <cerrno> // EBUSY +#include <pthread.h> + +#include "jexception.hpp" +#include "smutex.hpp" + +namespace mrg +{ +namespace journal2 +{ + + // Scoped-mutex (sm) container, superclass for scoped lock classes + class sm_cntnr + { + protected: + const smutex& _sm; + public: + sm_cntnr(const smutex& sm) : _sm(sm) {} + inline const smutex& get() const { return _sm; } + }; + + // Ultra-simple scoped lock class, auto-releases mutex when it goes out-of-scope + class slock : public sm_cntnr + { + public: + inline slock(const smutex& sm) : sm_cntnr(sm) + { + PTHREAD_CHK(::pthread_mutex_lock(_sm.get()), "::pthread_mutex_lock", "slock", "slock"); + } + virtual inline ~slock() + { + PTHREAD_CHK(::pthread_mutex_unlock(_sm.get()), "::pthread_mutex_unlock", "slock", "~slock"); + } + }; + + // Ultra-simple scoped try-lock class, auto-releases mutex when it goes out-of-scope + class stlock : public sm_cntnr + { + protected: + bool _locked; + public: + inline stlock(const smutex& sm) : sm_cntnr(sm), _locked(false) + { + int ret = ::pthread_mutex_trylock(_sm.get()); + _locked = (ret == 0); // check if lock obtained + if (!_locked && ret != EBUSY) PTHREAD_CHK(ret, "::pthread_mutex_trylock", "stlock", "stlock"); + } + virtual inline ~stlock() + { + if (_locked) + PTHREAD_CHK(::pthread_mutex_unlock(_sm.get()), "::pthread_mutex_unlock", "stlock", "~stlock"); + } + inline bool locked() const { return _locked; } + }; + +} // namespace journal2 +} // namespace mrg + +#endif // ifndef mrg_journal2_slock_hpp Added: store/trunk/cpp/lib/jrnl2/smutex.cpp =================================================================== --- store/trunk/cpp/lib/jrnl2/smutex.cpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/smutex.cpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,32 @@ +/** + * \file smutex.cpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#include "smutex.hpp" Added: store/trunk/cpp/lib/jrnl2/smutex.hpp =================================================================== --- store/trunk/cpp/lib/jrnl2/smutex.hpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/smutex.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,64 @@ +/** + * \file smutex.hpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#ifndef mrg_journal2_smutex_hpp +#define mrg_journal2_smutex_hpp + +#include <pthread.h> + +#include "jexception.hpp" + +namespace mrg +{ +namespace journal2 +{ + + // Ultra-simple scoped mutex class that allows a posix mutex to be initialized and destroyed with error checks + class smutex + { + protected: + mutable pthread_mutex_t _m; + public: + inline smutex() + { + PTHREAD_CHK(::pthread_mutex_init(&_m, 0), "::pthread_mutex_init", "smutex", "smutex"); + } + inline virtual ~smutex() + { + PTHREAD_CHK(::pthread_mutex_destroy(&_m), "::pthread_mutex_destroy", "smutex", "~smutex"); + } + inline pthread_mutex_t* get() const { return &_m; } + }; + +} // namespace journal2 +} // namespace mrg + +#endif // ifndef mrg_journal2_smutex_hpp Added: store/trunk/cpp/lib/jrnl2/store_params.cpp =================================================================== --- store/trunk/cpp/lib/jrnl2/store_params.cpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/store_params.cpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,115 @@ +/** + * \file store_params.cpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#include "store_params.hpp" + +#include <sstream> + +namespace mrg +{ +namespace journal2 +{ + +// static declarations +std::string store_params::_default_jid = "default_store"; +std::string store_params::_default_jrnl_dir = "/tmp/store"; +std::string store_params::_default_jrnl_base_filename = "JournalData"; +u_int16_t store_params::_default_num_jfiles = 8; +u_int32_t store_params::_default_jfsize_sblks = 3072; +bool store_params::_default_auto_expand = false; +u_int16_t store_params::_default_ae_max_jfiles = 0; +u_int16_t store_params::_default_wcache_num_pages = 32; +u_int32_t store_params::_default_wcache_pgsize_sblks = 128; + +store_params::store_params() : + _jrnl_id(_default_jid), + _jrnl_dir(_default_jrnl_dir), + _jrnl_base_filename(_default_jrnl_base_filename), + _num_jfiles(_default_num_jfiles), + _jfsize_sblks(_default_jfsize_sblks), + _auto_expand(_default_auto_expand), + _ae_max_jfiles(_default_ae_max_jfiles), + _wcache_num_pages(_default_wcache_num_pages), + _wcache_pgsize_sblks(_default_wcache_pgsize_sblks) +{} + +store_params::store_params(const std::string& jrnl_id, + const std::string& jrnl_dir, + const std::string& jrnl_base_filename, + const u_int16_t num_jfiles, + const u_int32_t jfsize_sblks, + const bool auto_expand, + const u_int16_t ae_max_jfiles, + const u_int16_t wcache_num_pages, + const u_int32_t wcache_pgsize_sblks) : + _jrnl_id(jrnl_id), + _jrnl_dir(jrnl_dir), + _jrnl_base_filename(jrnl_base_filename), + _num_jfiles(num_jfiles), + _jfsize_sblks(jfsize_sblks), + _auto_expand(auto_expand), + _ae_max_jfiles(ae_max_jfiles), + _wcache_num_pages(wcache_num_pages), + _wcache_pgsize_sblks(wcache_pgsize_sblks) +{} + +store_params::store_params(const store_params& sp) : + _jrnl_id(sp._jrnl_id), + _jrnl_dir(sp._jrnl_dir), + _jrnl_base_filename(sp._jrnl_base_filename), + _num_jfiles(sp._num_jfiles), + _jfsize_sblks(sp._jfsize_sblks), + _auto_expand(sp._auto_expand), + _ae_max_jfiles(sp._ae_max_jfiles), + _wcache_num_pages(sp._wcache_num_pages), + _wcache_pgsize_sblks(sp._wcache_pgsize_sblks) +{} + + +std::string +store_params::to_string() +{ + std::ostringstream oss; + oss << "Store Parameters:" << std::endl; + oss << " jrnl_id = \"" << _jrnl_id << "\"" << std::endl; + oss << " jrnl_dir = \"" << _jrnl_dir << "\"" << std::endl; + oss << " jrnl_base_filename = \"" << _jrnl_base_filename << "\"" << std::endl; + oss << " num_jfiles = " << _num_jfiles << std::endl; + oss << " jfsize_sblks = " << _jfsize_sblks << std::endl; + oss << " auto_expand = " << _auto_expand << std::endl; + oss << " ae_max_jfiles = " << _ae_max_jfiles << std::endl; + oss << " wcache_num_pages = " << _wcache_num_pages << std::endl; + oss << " wcache_pgsize_sblks = " << _wcache_pgsize_sblks << std::endl; + return oss.str(); +} + +} // namespace journal2 +} // namespace mrg Added: store/trunk/cpp/lib/jrnl2/store_params.hpp =================================================================== --- store/trunk/cpp/lib/jrnl2/store_params.hpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/store_params.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,82 @@ +/** + * \file store_params.hpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#ifndef mrg_journal2_store_params_hpp +#define mrg_journal2_store_params_hpp + +#include <sys/types.h> // u_int16_t, u_int32_t +#include <string> + +namespace mrg +{ +namespace journal2 +{ + + struct store_params + { + // static default store params + static std::string _default_jid; + static std::string _default_jrnl_dir; + static std::string _default_jrnl_base_filename; + static u_int16_t _default_num_jfiles; + static u_int32_t _default_jfsize_sblks; + static bool _default_auto_expand; + static u_int16_t _default_ae_max_jfiles; + static u_int16_t _default_wcache_num_pages; + static u_int32_t _default_wcache_pgsize_sblks; + + std::string _jrnl_id; + std::string _jrnl_dir; + std::string _jrnl_base_filename; + u_int16_t _num_jfiles; + u_int32_t _jfsize_sblks; + bool _auto_expand; + u_int16_t _ae_max_jfiles; + u_int16_t _wcache_num_pages; + u_int32_t _wcache_pgsize_sblks; + store_params(); + store_params(const std::string& jid, + const std::string& jrnl_dir, + const std::string& jrnl_base_filename, + const u_int16_t num_jfiles, + const u_int32_t jfsize_sblks, + const bool auto_expand, + const u_int16_t ae_max_jfiles, + const u_int16_t wcache_num_pages, + const u_int32_t wcache_pgsize_sblks); + store_params(const store_params& sp); + std::string to_string(); + }; + +} // namespace journal2 +} // namespace mrg + +#endif // mrg_journal2_store_params_hpp Added: store/trunk/cpp/lib/jrnl2/txn_ctxt.hpp =================================================================== --- store/trunk/cpp/lib/jrnl2/txn_ctxt.hpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/txn_ctxt.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,53 @@ +/** + * \file txn_ctxt.hpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#ifndef mrg_test_txn_ctxt_hpp +#define mrg_test_txn_ctxt_hpp + +#include <string> + +namespace mrg +{ +namespace journal2 +{ + + class txn_ctxt + { + public: + virtual std::string& tid() = 0; + virtual bool is_dist_txn() = 0; + }; + +} // namespace journal2 +} // namespace mrg + + +#endif // mrg_test_txn_ctxt_hpp Copied: store/trunk/cpp/perf/JournalInstance.cpp (from rev 4418, store/trunk/cpp/perf/perf_test.cpp) =================================================================== --- store/trunk/cpp/perf/JournalInstance.cpp (rev 0) +++ store/trunk/cpp/perf/JournalInstance.cpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,252 @@ +/** + * \file JournalInstance.cpp + * + * Qpid asynchronous store plugin library + * + * This file contains performance test code for the journal. + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#include "JournalInstance.hpp" + +#include <iostream> + +namespace mrg +{ +namespace test +{ + +JournalInstance::JournalInstance(const uint32_t num_msgs, + const uint32_t msg_size, + const char* msg_data, +#ifdef JOURNAL2 + mrg::journal2::jrnl* const jp) : +#else + mrg::journal::jcntl* const jp) : +#endif + _num_msgs(num_msgs), + _msg_size(msg_size), + _msg_data(msg_data), + _jp(jp), + _thread_switch(false) +{} + +JournalInstance::~JournalInstance() +{ + delete _jp; +} + + +// *** MUST BE THREAD-SAFE **** +// This method will be called by multiple threads simultaneously +// Enqueue thread entry point +void +JournalInstance::_enq() +{ + bool misfireFlag = false; + uint32_t i = 0; + while (i < _num_msgs) { +#ifdef JOURNAL2 + mrg::journal2::dtok* dtokp = new mrg::journal2::dtok(); + mrg::journal2::iores res = _jp->enqueue(_msg_data, _msg_size, dtokp); +#else + mrg::journal::data_tok* dtokp = new mrg::journal::data_tok(); + mrg::journal::iores res = _jp->enqueue_data_record(_msg_data, _msg_size, _msg_size, dtokp); +#endif + switch (res) { +#ifdef JOURNAL2 + case 0: +#else + case mrg::journal::RHM_IORES_SUCCESS: +#endif + i++; + misfireFlag = false; + break; +#ifdef JOURNAL2 + case mrg::journal2::RHM_IORES_BUSY: +#else + case mrg::journal::RHM_IORES_BUSY: +#endif + if (!misfireFlag) std::cout << "-" << std::flush; + delete dtokp; + misfireFlag = true; + break; +#ifdef JOURNAL2 + case mrg::journal2::RHM_IORES_ENQCAPTHRESH: +#else + case mrg::journal::RHM_IORES_ENQCAPTHRESH: +#endif + //std::cout << "_enq() RHM_IORES_ENQCAPTHRESH: " << dtokp->status_str() << std::endl; + if (!misfireFlag) std::cout << "*" << std::flush; + //std::cout << "."; + delete dtokp; + misfireFlag = true; + ::usleep(10); + break; + default: + delete dtokp; +#ifdef JOURNAL2 + std::cerr << "enqueue_data_record FAILED with " << mrg::journal2::iores_str(res) << std::endl; +#else + std::cerr << "enqueue_data_record FAILED with " << mrg::journal::iores_str(res) << std::endl; +#endif + } + } + _jp->flush(false); +} + + +// *** MUST BE THREAD-SAFE **** +// This method will be called by multiple threads simultaneously +// Dequeue thread entry point +void +JournalInstance::_deq() +{ + uint32_t i = 0; + while (i < _num_msgs) { +#ifdef JOURNAL2 + mrg::journal2::dtok* dtokp = 0; +#else + mrg::journal::data_tok* dtokp = 0; +#endif + while (!dtokp) { + std::size_t s; + bool get_events_flag; + { + std::unique_lock<std::mutex> l(_ucl_mutex); + get_events_flag = _ucl.size() == 0; + if (!get_events_flag) { + dtokp = _ucl.front(); + _ucl.pop(); + } + } + if (get_events_flag) { + _jp->get_wr_events(0); + ::usleep(1); + } + } + bool done = false; + while (!done) { +#ifdef JOURNAL2 + mrg::journal2::iores res = _jp->dequeue(dtokp); +#else + mrg::journal::iores res = _jp->dequeue_data_record(dtokp); +#endif + switch (res) { +#ifdef JOURNAL2 + case 0: +#else + case mrg::journal::RHM_IORES_SUCCESS: +#endif + i ++; + done = true; + break; +#ifdef JOURNAL2 + case mrg::journal2::RHM_IORES_BUSY: +#else + case mrg::journal::RHM_IORES_BUSY: +#endif + //::usleep(10); + break; + default: +#ifdef JOURNAL2 + std::cerr << "dequeue_data_record FAILED with " << mrg::journal2::iores_str(res) << ": " << dtokp->status_str() << std::endl; +#else + std::cerr << "dequeue_data_record FAILED with " << mrg::journal::iores_str(res) << ": " << dtokp->status_str() << std::endl; +#endif + delete dtokp; + done = true; + } + } + _jp->get_wr_events(0); + } + _jp->flush(true); +} + +// *** MUST BE THREAD-SAFE **** +// This method will be called by multiple threads simultaneously +// Main thread entry point. +// _thread_switch flips on each entry; this makes alternate threads run _enq() and _deq() +void +JournalInstance::operator()() +{ + bool ts; + { + std::lock_guard<std::mutex> l(_ts_lock); + ts = _thread_switch; + _thread_switch = !_thread_switch; + } + if (ts) { + _deq(); + } else { + _enq(); + } +} + +// *** MUST BE THREAD-SAFE **** +// This method will be called by multiple threads simultaneously +void +#ifdef JOURNAL2 +JournalInstance::wr_aio_cb(std::vector<mrg::journal2::dtok*>& dtokl) +{ + mrg::journal2::dtok* dtokp; +#else +JournalInstance::wr_aio_cb(std::vector<mrg::journal::data_tok*>& dtokl) +{ + mrg::journal::data_tok* dtokp; +#endif + while (dtokl.size()) + { + dtokp = dtokl.back(); + dtokl.pop_back(); +#ifdef JOURNAL2 + switch (dtokp->get_dtok_state().get_op_state()) + { + case mrg::journal2::OP_ENQUEUE: + +#else + switch (dtokp->wstate()) + { + case mrg::journal::data_tok::ENQ: +#endif + { + std::unique_lock<std::mutex> l(_ucl_mutex); + _ucl.push(dtokp); + } + break; + default: + delete dtokp; + } + } +} + +// *** MUST BE THREAD-SAFE **** +// This method will be called by multiple threads simultaneously +void +JournalInstance::rd_aio_cb(std::vector<u_int16_t>& pil) +{} + +} // namespace test +} // namespace mrg Copied: store/trunk/cpp/perf/JournalInstance.hpp (from rev 4418, store/trunk/cpp/perf/perf_test.hpp) =================================================================== --- store/trunk/cpp/perf/JournalInstance.hpp (rev 0) +++ store/trunk/cpp/perf/JournalInstance.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,95 @@ +/** + * \file JournalInstance.hpp + * + * Qpid asynchronous store plugin library + * + * This file contains performance test code for the journal. + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#ifndef mrg_test_JournalInstance_hpp +#define mrg_test_JournalInstance_hpp + +#include <cstdint> +#include <queue> +#include <mutex> + +#ifdef JOURNAL2 +#include "jrnl2/aio_callback.hpp" +#include "jrnl2/dtok.hpp" +#include "jrnl2/jrnl.hpp" +#else +#include "jrnl/aio_callback.hpp" +#include "jrnl/data_tok.hpp" +#include "jrnl/jcntl.hpp" +#endif + +namespace mrg +{ +namespace test +{ + +#ifdef JOURNAL2 + class JournalInstance: public mrg::journal2::aio_callback +#else + class JournalInstance: public mrg::journal::aio_callback +#endif + { + const uint32_t _num_msgs; + const uint32_t _msg_size; + const char* _msg_data; +#ifdef JOURNAL2 + mrg::journal2::jrnl* const _jp; + std::queue<mrg::journal2::dtok*> _ucl; +#else + mrg::journal::jcntl* const _jp; + std::queue<mrg::journal::data_tok*> _ucl; +#endif + std::mutex _ucl_mutex; + bool _thread_switch; + + std::mutex _ts_lock; + void _enq(); + void _deq(); + public: +#ifdef JOURNAL2 + JournalInstance(const uint32_t num_msgs, const uint32_t msg_size, const char* msg_data, mrg::journal2::jrnl* const jp); +#else + JournalInstance(const uint32_t num_msgs, const uint32_t msg_size, const char* msg_data, mrg::journal::jcntl* const jp); +#endif + virtual ~JournalInstance(); + void operator()(); +#ifdef JOURNAL2 + virtual void wr_aio_cb(std::vector<mrg::journal2::dtok*>& dtokl); +#else + virtual void wr_aio_cb(std::vector<mrg::journal::data_tok*>& dtokl); +#endif + virtual void rd_aio_cb(std::vector<u_int16_t>& pil); + }; + +} // namespace test +} // namespace mrg + +#endif // mrg_test_JournalInstance_hpp Copied: store/trunk/cpp/perf/JournalParameters.cpp (from rev 4418, store/trunk/cpp/perf/store_params.cpp) =================================================================== --- store/trunk/cpp/perf/JournalParameters.cpp (rev 0) +++ store/trunk/cpp/perf/JournalParameters.cpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,115 @@ +/** + * \file JournalParameters.cpp + * + * Qpid asynchronous store plugin library + * + * This file contains performance test code for the journal. + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#include "JournalParameters.hpp" + +#include <sstream> + +namespace mrg +{ +namespace test +{ + +// static declarations +std::string JournalParameters::_default_jid = "default_store"; +std::string JournalParameters::_default_jrnl_dir = "/tmp/store"; +std::string JournalParameters::_default_jrnl_base_filename = "JournalData"; +uint16_t JournalParameters::_default_num_jfiles = 8; +uint32_t JournalParameters::_default_jfsize_sblks = 3072; +bool JournalParameters::_default_auto_expand = false; +uint16_t JournalParameters::_default_ae_max_jfiles = 0; +uint16_t JournalParameters::_default_wcache_num_pages = 32; +uint32_t JournalParameters::_default_wcache_pgsize_sblks = 128; + +JournalParameters::JournalParameters() : + _jrnl_id(_default_jid), + _jrnl_dir(_default_jrnl_dir), + _jrnl_base_filename(_default_jrnl_base_filename), + _num_jfiles(_default_num_jfiles), + _jfsize_sblks(_default_jfsize_sblks), + _auto_expand(_default_auto_expand), + _ae_max_jfiles(_default_ae_max_jfiles), + _wcache_num_pages(_default_wcache_num_pages), + _wcache_pgsize_sblks(_default_wcache_pgsize_sblks) +{} + +JournalParameters::JournalParameters(const std::string& jrnl_id, + const std::string& jrnl_dir, + const std::string& jrnl_base_filename, + const u_int16_t num_jfiles, + const bool auto_expand, + const u_int16_t ae_max_jfiles, + const u_int32_t jfsize_sblks, + const u_int16_t wcache_num_pages, + const u_int32_t wcache_pgsize_sblks) : + _jrnl_id(jrnl_id), + _jrnl_dir(jrnl_dir), + _jrnl_base_filename(jrnl_base_filename), + _num_jfiles(num_jfiles), + _auto_expand(auto_expand), + _ae_max_jfiles(ae_max_jfiles), + _jfsize_sblks(jfsize_sblks), + _wcache_num_pages(wcache_num_pages), + _wcache_pgsize_sblks(wcache_pgsize_sblks) +{} + +JournalParameters::JournalParameters(const JournalParameters& sp) : + _jrnl_id(sp._jrnl_id), + _jrnl_dir(sp._jrnl_dir), + _jrnl_base_filename(sp._jrnl_base_filename), + _num_jfiles(sp._num_jfiles), + _auto_expand(sp._auto_expand), + _ae_max_jfiles(sp._ae_max_jfiles), + _jfsize_sblks(sp._jfsize_sblks), + _wcache_num_pages(sp._wcache_num_pages), + _wcache_pgsize_sblks(sp._wcache_pgsize_sblks) + +{} + +std::string +JournalParameters::to_string() +{ + std::ostringstream oss; + oss << "Store Parameters:" << std::endl; + oss << " jrnl_id = \"" << _jrnl_id << "\"" << std::endl; + oss << " jrnl_dir = \"" << _jrnl_dir << "\"" << std::endl; + oss << " jrnl_base_filename = \"" << _jrnl_base_filename << "\"" << std::endl; + oss << " num_jfiles = " << _num_jfiles << std::endl; + oss << " jfsize_sblks = " << _jfsize_sblks << std::endl; + oss << " auto_expand = " << _auto_expand << std::endl; + oss << " ae_max_jfiles = " << _ae_max_jfiles << std::endl; + oss << " wcache_num_pages = " << _wcache_num_pages << std::endl; + oss << " wcache_pgsize_sblks = " << _wcache_pgsize_sblks << std::endl; + return oss.str(); +} + +} // namespace test +} // namespace mrg Copied: store/trunk/cpp/perf/JournalParameters.hpp (from rev 4418, store/trunk/cpp/perf/store_params.hpp) =================================================================== --- store/trunk/cpp/perf/JournalParameters.hpp (rev 0) +++ store/trunk/cpp/perf/JournalParameters.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,82 @@ +/** + * \file JournalParameters.hpp + * + * Qpid asynchronous store plugin library + * + * This file contains performance test code for the journal. + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#ifndef mrg_test_JournalParameters_hpp +#define mrg_test_JournalParameters_hpp + +#include <cstdint> +#include <string> + +namespace mrg +{ +namespace test +{ + + struct JournalParameters + { + // static default store params + static std::string _default_jid; + static std::string _default_jrnl_dir; + static std::string _default_jrnl_base_filename; + static u_int16_t _default_num_jfiles; + static u_int32_t _default_jfsize_sblks; + static bool _default_auto_expand; + static u_int16_t _default_ae_max_jfiles; + static u_int16_t _default_wcache_num_pages; + static u_int32_t _default_wcache_pgsize_sblks; + + std::string _jrnl_id; + std::string _jrnl_dir; + std::string _jrnl_base_filename; + uint16_t _num_jfiles; + uint32_t _jfsize_sblks; + bool _auto_expand; + uint16_t _ae_max_jfiles; + uint16_t _wcache_num_pages; + uint32_t _wcache_pgsize_sblks; + JournalParameters(); + JournalParameters(const std::string& jid, + const std::string& jrnl_dir, + const std::string& jrnl_base_filename, + const uint16_t num_jfiles, + const bool auto_expand, + const uint16_t ae_max_jfiles, + const uint32_t jfsize_sblks, + const uint16_t wcache_num_pages, + const uint32_t wcache_pgsize_sblks); + JournalParameters(const JournalParameters& sp); + std::string to_string(); + }; + +} // namespace test +} // namespace mrg + +#endif // mrg_test_JournalParameters_hpp Modified: store/trunk/cpp/perf/Makefile.am =================================================================== --- store/trunk/cpp/perf/Makefile.am 2010-12-02 15:11:07 UTC (rev 4419) +++ store/trunk/cpp/perf/Makefile.am 2010-12-02 20:48:07 UTC (rev 4420) @@ -28,15 +28,15 @@ bin_PROGRAMS = perf perf_SOURCES = \ - perf.cpp \ - perf_test.cpp \ - perf_timer.cpp \ - store_params.cpp \ - test_params.cpp \ - perf.hpp \ - perf_test.hpp \ - perf_timer.hpp \ - store_params.hpp \ - test_params.hpp + JournalInstance.cpp \ + JournalParameters.cpp \ + PerformanceTimer.cpp \ + StorePerformanceTest.cpp \ + TestParameters.cpp \ + JournalInstance.hpp \ + JournalParameters.hpp \ + PerformanceTimer.hpp \ + StorePerformanceTest.hpp \ + TestParameters.hpp perf_CXXFLAGS = -std=c++0x -lpthread perf_LDADD = -lrt $(top_builddir)/lib/msgstore.la Copied: store/trunk/cpp/perf/PerformanceTimer.cpp (from rev 4418, store/trunk/cpp/perf/perf_timer.cpp) =================================================================== --- store/trunk/cpp/perf/PerformanceTimer.cpp (rev 0) +++ store/trunk/cpp/perf/PerformanceTimer.cpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,80 @@ +/** + * \file PerformanceTimer.cpp + * + * Qpid asynchronous store plugin library + * + * This file contains performance test code for the journal. + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#include "PerformanceTimer.hpp" + +#include <iostream> + +namespace mrg +{ +namespace test +{ + PerformanceTimer::PerformanceTimer(const uint32_t num_msgs, + const uint32_t msg_size, + const uint16_t num_queues, + const uint16_t num_threads_per_queue) : + _num_msgs(num_msgs), + _msg_size(msg_size), + _num_queues(num_queues), + _num_threads_per_queue(num_threads_per_queue) + { + ::timespec ts; + ::clock_gettime(CLOCK_REALTIME, &ts); + _start_time = get_double_time(ts); + } + + PerformanceTimer::~PerformanceTimer() + { + ::timespec ts; + ::clock_gettime(CLOCK_REALTIME, &ts); + double time_taken = get_double_time(ts) - _start_time; + std::cout << "TEST RESULTS:" << std::endl; + std::cout << " Msgs per thread: " << _num_msgs << std::endl; + std::cout << " Msg size: " << _msg_size << std::endl; + std::cout << " No. queues: " << _num_queues << std::endl; + std::cout << " No. threads/queue: " << _num_threads_per_queue << std::endl; + std::cout << " Time taken: " << time_taken << " sec" << std::endl; + u_int32_t msgs = _num_msgs * _num_queues * _num_threads_per_queue; + std::cout << " Total no. msgs: " << msgs << std::endl; + double msgs_per_sec = msgs / time_taken; + std::cout << " Msg throughput: " << (msgs_per_sec / 1e3) << " kMsgs/sec" << std::endl; + std::cout << " " << (msgs_per_sec * _msg_size / 1e6) << " MB/sec" << std::endl; + } + + // static + double PerformanceTimer::get_double_time(const ::timespec& ts) + { + return ts.tv_sec + (double(ts.tv_nsec) / 1e9); + } + + +} // namespace test +} // namespace mrg Copied: store/trunk/cpp/perf/PerformanceTimer.hpp (from rev 4418, store/trunk/cpp/perf/perf_timer.hpp) =================================================================== --- store/trunk/cpp/perf/PerformanceTimer.hpp (rev 0) +++ store/trunk/cpp/perf/PerformanceTimer.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,63 @@ +/** + * \file PerformanceTimer.hpp + * + * Qpid asynchronous store plugin library + * + * This file contains performance test code for the journal. + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#ifndef mrg_test_PerformanceTimer_hpp +#define mrg_test_PerformanceTimer_hpp + +#include <cstdint> +#include <ctime> + +namespace mrg +{ +namespace test +{ + + class PerformanceTimer + { + protected: + uint32_t _num_msgs; + uint32_t _msg_size; + uint16_t _num_queues; + uint16_t _num_threads_per_queue; + double _start_time; + static double get_double_time(const timespec& tx); + public: + PerformanceTimer(const uint32_t num_msgs, + const uint32_t msg_size, + const uint16_t num_queues, + const uint16_t num_threads_per_queue); + ~PerformanceTimer(); + }; + +} // namespace test +} // namespace mrg + +#endif // mrg_test_PerformanceTimer_hpp Copied: store/trunk/cpp/perf/StorePerformanceTest.cpp (from rev 4418, store/trunk/cpp/perf/perf.cpp) =================================================================== --- store/trunk/cpp/perf/StorePerformanceTest.cpp (rev 0) +++ store/trunk/cpp/perf/StorePerformanceTest.cpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,313 @@ +/** + * \file StorePerformanceTest.cpp + * + * Qpid asynchronous store plugin library + * + * This file contains performance test code for the journal. + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#include "StorePerformanceTest.hpp" + +#include <cstdint> // uint16_t, uint32_t +#include <cstdlib> // atof, atoi, atol +#include <deque> +#include <iomanip> +#include <iostream> +#include <sstream> +#include <string> +#include <thread> + +#include <getopt.h> +#include <unistd.h> + +#include "PerformanceTimer.hpp" + +#ifdef JOURNAL2 +#include "jrnl2/jdir.hpp" +#else +#include "jrnl/jdir.hpp" +#endif + +namespace mrg +{ +namespace test +{ + +#ifdef JOURNAL2 +StorePerformanceTest::StorePerformanceTest(const TestParameters& tp, const mrg::journal2::store_params& sp) : +#else +StorePerformanceTest::StorePerformanceTest(const TestParameters& tp, const JournalParameters& sp) : +#endif + _test_params(tp), + _store_params(sp), + _msg_data(new char[tp._msg_size]) +{} + +StorePerformanceTest::~StorePerformanceTest() +{ + while (_tests.size()) + { + delete _tests.back(); + _tests.pop_back(); + } + delete[] _msg_data; +} + +void +StorePerformanceTest::_prepare_journals() +{ +#ifdef JOURNAL2 + if (mrg::journal2::jdir::exists(_store_params._jrnl_dir)) mrg::journal2::jdir::delete_dir(_store_params._jrnl_dir); + mrg::journal2::jdir::create_dir(_store_params._jrnl_dir); + mrg::journal2::jrnl* jp; +#else + if (mrg::journal::jdir::exists(_store_params._jrnl_dir)) mrg::journal::jdir::delete_dir(_store_params._jrnl_dir); + mrg::journal::jdir::create_dir(_store_params._jrnl_dir); + mrg::journal::jcntl* jp; +#endif + JournalInstance* ptp; + for (uint16_t j = 0; j < _test_params._num_queues; j++) + { + std::ostringstream jname; + jname << "jrnl_" << std::setw(4) << std::setfill('0') << j; + std::ostringstream jdir; + jdir << _store_params._jrnl_dir << "/" << jname.str(); +#ifdef JOURNAL2 + jp = new mrg::journal2::jrnl(jname.str(), jdir.str(), _store_params._jrnl_base_filename); +#else + jp = new mrg::journal::jcntl(jname.str(), jdir.str(), _store_params._jrnl_base_filename); +#endif + ptp = new JournalInstance(_test_params._num_msgs, _test_params._msg_size, _msg_data, jp); +#ifdef JOURNAL2 + jp->initialize(&_store_params, ptp); +#else + jp->initialize(_store_params._num_jfiles, _store_params._auto_expand, _store_params._ae_max_jfiles, + _store_params._jfsize_sblks, _store_params._wcache_num_pages, + _store_params._wcache_pgsize_sblks, ptp); +#endif + + _tests.push_back(ptp); + } +} + +void +StorePerformanceTest::run() +{ + std::deque<std::thread*> threads; + std::thread* tp; + _prepare_journals(); + { + // --- Start of timed section --- + mrg::test::PerformanceTimer t(_test_params._num_msgs, _test_params._msg_size, _test_params._num_queues, + _test_params._num_thread_pairs_per_queue); + + for (uint16_t q=0; q<_test_params._num_queues; q++) + { + // Launch threads in pairs + for (uint16_t t=0; t<_test_params._num_thread_pairs_per_queue * 2; t++) + { + tp = new std::thread(std::ref(*_tests[q])); + threads.push_back(tp); + } + } + + while (threads.size()) + { + threads.front()->join(); + delete threads.front(); + threads.pop_front(); + } + std::cout << std::endl; + // --- End of timed section --- + } +} + +void +print_args() +{ + std::cout << " -h --help: This help message" << std::endl; + std::cout << std::endl; + std::cout << "Test params:" << std::endl; + std::cout << " -M --num_msgs: Number of messages to send [" + << TestParameters::_default_num_msgs << "]" << std::endl; + std::cout << " -S --msg_size: Size of each message to be sent [" + << TestParameters::_default_msg_size << "]" << std::endl; + std::cout << " -Q --num_queues: Number of simultaneous queues [" + << TestParameters::_default_num_queues << "]" << std::endl; + std::cout << " -T --num_thread_pairs_per_queue: Number of thread pairs per queue [" + << TestParameters::_default_num_thread_pairs_per_queue << "]" << std::endl; + std::cout << " -E --enq_txn_blk_size: Enqueue transaction block size (0=non-txn) [" + << TestParameters::_default_enq_txn_blk_size << "]" << std::endl; + std::cout << " -D --deq_txn_blk_size: Dequeue transaction block size (0=non-txn) [" + << TestParameters::_default_deq_txn_blk_size << "]" << std::endl; + std::cout << std::endl; + std::cout << "Store params:" << std::endl; +#ifdef JOURNAL2 + std::cout << " -d --jrnl_dir: Store directory [\"" + << mrg::journal2::store_params::_default_jrnl_dir << "\"]" << std::endl; + std::cout << " -b --jrnl_base_filename: Base name for journal files [\"" + << mrg::journal2::store_params::_default_jrnl_base_filename << "\"]" << std::endl; + std::cout << " -f --num_jfiles: Number of journal files [" + << mrg::journal2::store_params::_default_num_jfiles << "]" << std::endl; + std::cout << " -s --jfsize_sblks: Size of each journal file in sblks (512 byte blocks) [" + << mrg::journal2::store_params::_default_jfsize_sblks << "]" << std::endl; + std::cout << " -a --auto_expand: Auto-expand the journal [" + << (mrg::journal2::store_params::_default_auto_expand?"T":"F") << "]" << std::endl; + std::cout << " -e --ae_max_jfiles: Upper limit on number of auto-expanded journal files [" + << mrg::journal2::store_params::_default_ae_max_jfiles << "]" << std::endl; + std::cout << " -p --wcache_num_pages: Number of write buffer pages [" + << mrg::journal2::store_params::_default_wcache_num_pages << "]" << std::endl; + std::cout << " -c --wcache_pgsize_sblks: Size of each write buffer page in sblks (512 byte blocks) [" + << mrg::journal2::store_params::_default_wcache_pgsize_sblks << "]" << std::endl; +#else + std::cout << " -d --jrnl_dir: Store directory [\"" + << JournalParameters::_default_jrnl_dir << "\"]" << std::endl; + std::cout << " -b --jrnl_base_filename: Base name for journal files [\"" + << JournalParameters::_default_jrnl_base_filename << "\"]" << std::endl; + std::cout << " -f --num_jfiles: Number of journal files [" + << JournalParameters::_default_num_jfiles << "]" << std::endl; + std::cout << " -s --jfsize_sblks: Size of each journal file in sblks (512 byte blocks) [" + << JournalParameters::_default_jfsize_sblks << "]" << std::endl; + std::cout << " -a --auto_expand: Auto-expand the journal [" + << (JournalParameters::_default_auto_expand?"T":"F") << "]" << std::endl; + std::cout << " -e --ae_max_jfiles: Upper limit on number of auto-expanded journal files [" + << JournalParameters::_default_ae_max_jfiles << "]" << std::endl; + std::cout << " -p --wcache_num_pages: Number of write buffer pages [" + << JournalParameters::_default_wcache_num_pages << "]" << std::endl; + std::cout << " -c --wcache_pgsize_sblks: Size of each write buffer page in sblks (512 byte blocks) [" + << JournalParameters::_default_wcache_pgsize_sblks << "]" << std::endl; +#endif +} + +bool +#ifdef JOURNAL2 +read_args(int argc, char** argv, mrg::test::TestParameters& tp, mrg::journal2::store_params& sp) +#else +read_args(int argc, char** argv, mrg::test::TestParameters& tp, mrg::test::JournalParameters& sp) +#endif +{ + static struct option long_options[] = { + {"help", no_argument, 0, 'h'}, + + // Test params + {"num_msgs", required_argument, 0, 'm'}, + {"msg_size", required_argument, 0, 'S'}, + {"num_queues", required_argument, 0, 'q'}, + {"num_threads_per_queue", required_argument, 0, 't'}, + + // Journal params + {"jrnl_dir", required_argument, 0, 'd'}, + {"jrnl_base_filename", required_argument, 0, 'b'}, + {"num_jfiles", required_argument, 0, 'f'}, + {"jfsize_sblks", required_argument, 0, 's'}, + {"auto_expand", no_argument, 0, 'a'}, + {"ae_max_jfiles", required_argument, 0, 'e'}, + {"wcache_num_pages", required_argument, 0, 'p'}, + {"wcache_pgsize_sblks", required_argument, 0, 'c'}, + + {0, 0, 0, 0} + }; + bool err = false; + int c = 0; + while (true) + { + int option_index = 0; + c = getopt_long(argc, argv, "ab:c:d:e:f:hm:p:q:s:S:t:", long_options, &option_index); + if (c == -1) break; + switch (c) + { + // Test params + case 'm': + tp._num_msgs = uint32_t(std::atol(optarg)); + break; + case 'S': + tp._msg_size = uint32_t(std::atol(optarg)); + break; + case 'q': + tp._num_queues = uint16_t(std::atoi(optarg)); + break; + case 't': + tp._num_thread_pairs_per_queue = uint16_t(std::atoi(optarg)); + break; + + // Store params + case 'd': + sp._jrnl_dir.assign(optarg); + break; + case 'b': + sp._jrnl_base_filename.assign(optarg); + break; + case 'f': + sp._num_jfiles = uint16_t(std::atoi(optarg)); + break; + case 's': + sp._jfsize_sblks = uint32_t(std::atol(optarg)); + break; + case 'a': + sp._auto_expand = true; + break; + case 'e': + sp._ae_max_jfiles = uint16_t(std::atoi(optarg)); + break; + case 'p': + sp._wcache_num_pages = uint16_t(std::atoi(optarg)); + break; + case 'c': + sp._wcache_pgsize_sblks = uint32_t(std::atol(optarg)); + break; + + // Other + case 'h': + default: + err = true; + print_args(); + } + } + return err; +} + +} // namespace test +} // namespace mrg + +// ----------------------------------------------------------------- + +int +main(int argc, char** argv) +{ + mrg::test::TestParameters tp; +#ifdef JOURNAL2 + mrg::journal2::store_params sp; +#else + mrg::test::JournalParameters sp; +#endif + if (mrg::test::read_args(argc, argv, tp, sp)) return 1; + std::cout << tp.to_string() << std::endl; + std::cout << sp.to_string() << std::endl; + mrg::test::StorePerformanceTest p(tp, sp); + p.run(); + + return 0; +} Copied: store/trunk/cpp/perf/StorePerformanceTest.hpp (from rev 4418, store/trunk/cpp/perf/perf.hpp) =================================================================== --- store/trunk/cpp/perf/StorePerformanceTest.hpp (rev 0) +++ store/trunk/cpp/perf/StorePerformanceTest.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,84 @@ +/** + * \file StorePerformanceTest.hpp + * + * Qpid asynchronous store plugin library + * + * This file contains performance test code for the journal. + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#ifndef mrg_test_StorePerformanceTest_hpp +#define mrg_test_StorePerformanceTest_hpp + +#include <vector> + +#include "JournalInstance.hpp" +#include "TestParameters.hpp" + +#ifdef JOURNAL2 +#include "jrnl2/store_params.hpp" +#else +#include "JournalParameters.hpp" +#endif + + + +namespace mrg +{ +namespace test +{ + + class StorePerformanceTest + { + const TestParameters& _test_params; +#ifdef JOURNAL2 + const mrg::journal2::store_params& _store_params; +#else + const JournalParameters& _store_params; +#endif + const char* _msg_data; + std::vector<JournalInstance*> _tests; + void _prepare_journals(); + public: +#ifdef JOURNAL2 + StorePerformanceTest(const TestParameters& tp, const mrg::journal2::store_params& sp); +#else + StorePerformanceTest(const TestParameters& tp, const JournalParameters& sp); +#endif + ~StorePerformanceTest(); + void run(); + }; + + void print_args(); +#ifdef JOURNAL2 + bool read_args(int argc, char** argv, mrg::test::TestParameters& tp, mrg::journal2::store_params& sp); +#else + bool read_args(int argc, char** argv, mrg::test::TestParameters& tp, mrg::test::JournalParameters& sp); +#endif + +} // namespace test +} // namespace mrg + +#endif // mrg_test_StorePerformanceTest_hpp Copied: store/trunk/cpp/perf/TestParameters.cpp (from rev 4418, store/trunk/cpp/perf/test_params.cpp) =================================================================== --- store/trunk/cpp/perf/TestParameters.cpp (rev 0) +++ store/trunk/cpp/perf/TestParameters.cpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,96 @@ +/** + * \file TestParameters.cpp + * + * Qpid asynchronous store plugin library + * + * This file contains performance test code for the journal. + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#include "TestParameters.hpp" + +#include <sstream> + +namespace mrg +{ +namespace test +{ + +// static declarations +uint32_t TestParameters::_default_num_msgs = 100; +uint32_t TestParameters::_default_msg_size = 10; +uint16_t TestParameters::_default_num_queues = 1; +uint16_t TestParameters::_default_num_thread_pairs_per_queue = 1; +uint16_t TestParameters::_default_enq_txn_blk_size = 0; +uint16_t TestParameters::_default_deq_txn_blk_size = 0; + +TestParameters::TestParameters(): + _num_msgs(_default_num_msgs), + _msg_size(_default_msg_size), + _num_queues(_default_num_queues), + _num_thread_pairs_per_queue(_default_num_thread_pairs_per_queue), + _enq_txn_blk_size(_default_enq_txn_blk_size), + _deq_txn_blk_size(_default_deq_txn_blk_size) +{} + +TestParameters::TestParameters(const u_int32_t num_msgs, + const u_int32_t msg_size, + const u_int16_t num_queues, + const u_int16_t num_thread_pairs_per_queue, + const u_int16_t enq_txn_blk_size, + const u_int16_t deq_txn_blk_size) : + _num_msgs(num_msgs), + _msg_size(msg_size), + _num_queues(num_queues), + _num_thread_pairs_per_queue(num_thread_pairs_per_queue), + _enq_txn_blk_size(enq_txn_blk_size), + _deq_txn_blk_size(deq_txn_blk_size) +{} + +TestParameters::TestParameters(const TestParameters& tp): + _num_msgs(tp._num_msgs), + _msg_size(tp._msg_size), + _num_queues(tp._num_queues), + _num_thread_pairs_per_queue(tp._num_thread_pairs_per_queue), + _enq_txn_blk_size(tp._enq_txn_blk_size), + _deq_txn_blk_size(tp._deq_txn_blk_size) +{} + +std::string +TestParameters::to_string() +{ + std::ostringstream oss; + oss << "Test Parameters:" << std::endl; + oss << " num_msgs = " << _num_msgs << std::endl; + oss << " msg_size = " << _msg_size << std::endl; + oss << " num_queues = " << _num_queues << std::endl; + oss << " num_thread_pairs_per_queue = " << _num_thread_pairs_per_queue << std::endl; + oss << " enq_txn_blk_size = " << _enq_txn_blk_size << std::endl; + oss << " deq_txn_blk_size = " << _deq_txn_blk_size << std::endl; + return oss.str(); +} + +} // namespace test +} // namespace mrg Copied: store/trunk/cpp/perf/TestParameters.hpp (from rev 4418, store/trunk/cpp/perf/test_params.hpp) =================================================================== --- store/trunk/cpp/perf/TestParameters.hpp (rev 0) +++ store/trunk/cpp/perf/TestParameters.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,72 @@ +/** + * \file TestParameters.hpp + * + * Qpid asynchronous store plugin library + * + * This file contains performance test code for the journal. + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#ifndef mrg_test_TestParameters_hpp +#define mrg_test_TestParameters_hpp + +#include <cstdint> +#include <string> + +namespace mrg +{ +namespace test +{ + + struct TestParameters + { + static uint32_t _default_num_msgs; + static uint32_t _default_msg_size; + static uint16_t _default_num_queues; + static uint16_t _default_num_thread_pairs_per_queue; + static uint16_t _default_enq_txn_blk_size; + static uint16_t _default_deq_txn_blk_size; + + uint32_t _num_msgs; + uint32_t _msg_size; + uint16_t _num_queues; + uint16_t _num_thread_pairs_per_queue; + uint16_t _enq_txn_blk_size; + uint16_t _deq_txn_blk_size; + TestParameters(); + TestParameters(const uint32_t num_msgs, + const uint32_t msg_size, + const uint16_t num_queues, + const uint16_t num_thread_pairs_per_queue, + const uint16_t enq_txn_blk_size, + const uint16_t deq_txn_blk_size); + TestParameters(const TestParameters& tp); + std::string to_string(); + }; + +} // namespace test +} // namespace mrg + +#endif // mrg_test_TestParameters_hpp Added: store/trunk/cpp/perf/m =================================================================== --- store/trunk/cpp/perf/m (rev 0) +++ store/trunk/cpp/perf/m 2010-12-02 20:48:07 UTC (rev 4420) @@ -0,0 +1,37 @@ +#!/bin/bash + +# This script builds an executable 'perf' directly from the source files. This is suitable for testing +# using valgrind and similar tools which don't play well with libtool. + +# The variable JOURNAL2, if defined, will link with the new journal2 namespace journal. Otherwise the old journal +# namespace will be used. +#JOURNAL2=1 + +# Optimization options +#OPT="-O0 -ggdb" +OPT="-O3 -g0 -DNDEBUG" + +PERF_FILES="StorePerformanceTest.cpp JournalInstance.cpp PerformanceTimer.cpp JournalParameters.cpp TestParameters.cpp" + +if [[ ${JOURNAL2}x == x ]] ; then + +JRNL_FILES="../lib/jrnl/aio.cpp ../lib/jrnl/enq_map.cpp ../lib/jrnl/jdir.cpp ../lib/jrnl/jrec.cpp ../lib/jrnl/rfc.cpp \ + ../lib/jrnl/smutex.cpp ../lib/jrnl/wmgr.cpp ../lib/jrnl/cvar.cpp ../lib/jrnl/enq_rec.cpp ../lib/jrnl/jerrno.cpp \ + ../lib/jrnl/lp_map.cpp ../lib/jrnl/rmgr.cpp ../lib/jrnl/time_ns.cpp ../lib/jrnl/wrfc.cpp ../lib/jrnl/data_tok.cpp \ + ../lib/jrnl/fcntl.cpp ../lib/jrnl/jexception.cpp ../lib/jrnl/lpmgr.cpp ../lib/jrnl/rrfc.cpp ../lib/jrnl/txn_map.cpp \ + ../lib/jrnl/deq_rec.cpp ../lib/jrnl/jcntl.cpp ../lib/jrnl/jinf.cpp ../lib/jrnl/pmgr.cpp ../lib/jrnl/slock.cpp \ + ../lib/jrnl/txn_rec.cpp" + +else + +DEFINES=-DJOURNAL2 + +JRNL_FILES="../lib/jrnl2/dtok.cpp ../lib/jrnl2/jdir.cpp ../lib/jrnl2/jexception.cpp ../lib/jrnl2/jrnl_state.cpp \ + ../lib/jrnl2/smutex.cpp ../lib/jrnl2/dtok_state.cpp ../lib/jrnl2/jerrno.cpp ../lib/jrnl2/jrnl.cpp \ + ../lib/jrnl2/slock.cpp ../lib/jrnl2/store_params.cpp" + +fi + +rm -f *.o perf +echo g++ -o perf -I ../lib -std=c++0x ${OPT} -lrt -laio -lpthread ${DEFINES} ${PERF_FILES} ${JRNL_FILES} +g++ -o perf -I ../lib -std=c++0x ${OPT} -lrt -laio -lpthread ${DEFINES} ${PERF_FILES} ${JRNL_FILES} Deleted: store/trunk/cpp/perf/perf.cpp =================================================================== --- store/trunk/cpp/perf/perf.cpp 2010-12-02 15:11:07 UTC (rev 4419) +++ store/trunk/cpp/perf/perf.cpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -1,248 +0,0 @@ -/** - * \file perf.cpp - * - * Qpid asynchronous store plugin library - * - * This file contains performance test code for the journal. - * - * \author Kim van der Riet - * - * Copyright (c) 2010 Red Hat, Inc. - * - * This file is part of the Qpid async store library msgstore.so. - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation; either - * version 2.1 of the License, or (at your option) any later version. - * - * This library is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 - * USA - * - * The GNU Lesser General Public License is available in the file COPYING. - */ - -#include "perf.hpp" - -#include <cstdint> // uint16_t, uint32_t -#include <cstdlib> // atof, atoi, atol -#include <deque> -#include <iostream> -#include <sstream> -#include <string> -#include <thread> - -#include <getopt.h> -#include <unistd.h> - -#include "jrnl/jdir.hpp" -#include "perf_timer.hpp" - -namespace mrg -{ -namespace test -{ - -perf::perf(const test_params& tp, const store_params& sp) : - _test_params(tp), - _store_params(sp), - _msg_data(new char[tp._msg_size]) -{} - -perf::~perf() -{ - while (_tests.size()) - { - delete _tests.back(); - _tests.pop_back(); - } - delete[] _msg_data; -} - -void -perf::prepare_journals() -{ - if (mrg::journal::jdir::exists(_store_params._jrnl_dir)) mrg::journal::jdir::delete_dir(_store_params._jrnl_dir); - mrg::journal::jdir::create_dir(_store_params._jrnl_dir); - - mrg::journal::jcntl* jp; - perf_test* ptp; - for (uint16_t j = 0; j < _test_params._num_queues; j++) - { - std::ostringstream jname; - jname << "jrnl_" << std::setw(4) << std::setfill('0') << j; - std::ostringstream jdir; - jdir << _store_params._jrnl_dir << "/" << jname.str(); - jp = new mrg::journal::jcntl(jname.str(), jdir.str(), _store_params._jrnl_base_filename); - ptp = new perf_test(_test_params._num_msgs, _test_params._msg_size, _msg_data, jp); - jp->initialize(_store_params._num_jfiles, _store_params._auto_expand, _store_params._ae_max_jfiles, - _store_params._jfsize_sblks, _store_params._wcache_num_pages, - _store_params._wcache_pgsize_sblks, ptp); - - _tests.push_back(ptp); - } -} - -void -perf::run() -{ - std::deque<std::thread*> threads; - std::thread* thread; - prepare_journals(); - { - // --- Start of timed section --- - mrg::test::perf_timer t(_test_params._num_msgs, _test_params._msg_size, _test_params._num_queues, - _test_params._num_threads_per_queue); - - for (uint16_t q=0; q<_test_params._num_queues; q++) - { - for (uint16_t t=0; t<_test_params._num_threads_per_queue; t++) - { - thread = new std::thread(std::ref(*_tests[q])); - threads.push_back(thread); - } - } - - while (threads.size()) - { - threads.front()->join(); - delete threads.front(); - threads.pop_front(); - } - // --- End of timed section --- - } -} - -} // namespace test -} // namespace mrg - -// ----------------------------------------------------------------- - -void -print_args() -{ - mrg::test::test_params tp; - mrg::test::store_params sp; - - std::cout << "-h --help: This help message" << std::endl; - std::cout << std::endl; - std::cout << "Test params:" << std::endl; - std::cout << "-m --num_msgs: Number of messages to send" << std::endl; - std::cout << "-S --msg_size: Size of each message to be sent" << std::endl; - std::cout << "-q --num_queues: Number of simultaneous queues" << std::endl; - std::cout << "-t --num_threads_per_queue: Number of threads per queue" << std::endl; - std::cout << std::endl; - std::cout << "Store params:" << std::endl; - std::cout << "-d --jrnl_dir: Store directory" << std::endl; - std::cout << "-b --jrnl_base_filename: Base name for journal files" << std::endl; - std::cout << "-f --num_jfiles: Number of journal files" << std::endl; - std::cout << "-s --jfsize_sblks: Size of each journal file in sblks (512 byte blocks)" << std::endl; - std::cout << "-a --auto_expand: Auto-expand the journal" << std::endl; - std::cout << "-e --ae_max_jfiles: Upper limit on number of auto-expanded journal files" << std::endl; - std::cout << "-p --wcache_num_pages: Number of write buffer pages" << std::endl; - std::cout << "-c --wcache_pgsize_sblks: Size of each write buffer page in sblks (512 byte blocks)" << std::endl; -} - -bool -read_args(int argc, char** argv, mrg::test::test_params& tp, mrg::test::store_params& sp) -{ - static struct option long_options[] = { - {"help", no_argument, 0, 'h'}, - - // Test params - {"num_msgs", required_argument, 0, 'm'}, - {"msg_size", required_argument, 0, 'S'}, - {"num_queues", required_argument, 0, 'q'}, - {"num_threads_per_queue", required_argument, 0, 't'}, - - // Journal params - {"jrnl_dir", required_argument, 0, 'd'}, - {"jrnl_base_filename", required_argument, 0, 'b'}, - {"num_jfiles", required_argument, 0, 'f'}, - {"jfsize_sblks", required_argument, 0, 's'}, - {"auto_expand", no_argument, 0, 'a'}, - {"ae_max_jfiles", required_argument, 0, 'e'}, - {"wcache_num_pages", required_argument, 0, 'p'}, - {"wcache_pgsize_sblks", required_argument, 0, 'c'}, - - {0, 0, 0, 0} - }; - bool err = false; - int c = 0; - while (true) - { - int option_index = 0; - c = getopt_long(argc, argv, "ab:c:d:e:f:hm:p:q:s:S:t:", long_options, &option_index); - if (c == -1) break; - switch (c) - { - // Test params - case 'm': - tp._num_msgs = uint32_t(std::atol(optarg)); - break; - case 'S': - tp._msg_size = uint32_t(std::atol(optarg)); - break; - case 'q': - tp._num_queues = uint16_t(std::atoi(optarg)); - break; - case 't': - tp._num_threads_per_queue = uint16_t(std::atoi(optarg)); - break; - - // Store params - case 'd': - sp._jrnl_dir.assign(optarg); - break; - case 'b': - sp._jrnl_base_filename.assign(optarg); - break; - case 'f': - sp._num_jfiles = uint16_t(std::atoi(optarg)); - break; - case 's': - sp._jfsize_sblks = uint32_t(std::atol(optarg)); - break; - case 'a': - sp._auto_expand = true; - break; - case 'e': - sp._ae_max_jfiles = uint16_t(std::atoi(optarg)); - break; - case 'p': - sp._wcache_num_pages = uint16_t(std::atoi(optarg)); - break; - case 'c': - sp._wcache_pgsize_sblks = uint32_t(std::atol(optarg)); - break; - - // Other - case 'h': - default: - err = true; - print_args(); - } - } - return err; -} - -int -main(int argc, char** argv) -{ - mrg::test::test_params tp; - mrg::test::store_params sp; - if (read_args(argc, argv, tp, sp)) return 1; - std::cout << tp.to_string() << std::endl; - std::cout << sp.to_string() << std::endl; - mrg::test::perf p(tp, sp); - p.run(); - - return 0; -} Deleted: store/trunk/cpp/perf/perf.hpp =================================================================== --- store/trunk/cpp/perf/perf.hpp 2010-12-02 15:11:07 UTC (rev 4419) +++ store/trunk/cpp/perf/perf.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -1,65 +0,0 @@ -/** - * \file perf.hpp - * - * Qpid asynchronous store plugin library - * - * This file contains performance test code for the journal. - * - * \author Kim van der Riet - * - * Copyright (c) 2010 Red Hat, Inc. - * - * This file is part of the Qpid async store library msgstore.so. - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation; either - * version 2.1 of the License, or (at your option) any later version. - * - * This library is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 - * USA - * - * The GNU Lesser General Public License is available in the file COPYING. - */ - -#ifndef mrg_test_perf_hpp -#define mrg_test_perf_hpp - -#include <vector> - -#include "perf_test.hpp" -#include "store_params.hpp" -#include "test_params.hpp" - -namespace mrg -{ -namespace test -{ - - class perf - { - const test_params& _test_params; - const store_params& _store_params; - const char* _msg_data; - std::vector<perf_test*> _tests; - void prepare_journals(); - public: - perf(const test_params& tp, const store_params& sp); - ~perf(); - void run(); - }; - -} // namespace test -} // namespace mrg - -void print_args(); -bool read_args(int argc, char** argv, mrg::test::test_params& tp, mrg::test::store_params& sp); - -#endif // mrg_test_perf_hpp Deleted: store/trunk/cpp/perf/perf_test.cpp =================================================================== --- store/trunk/cpp/perf/perf_test.cpp 2010-12-02 15:11:07 UTC (rev 4419) +++ store/trunk/cpp/perf/perf_test.cpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -1,198 +0,0 @@ -/** - * \file perf_test.cpp - * - * Qpid asynchronous store plugin library - * - * This file contains performance test code for the journal. - * - * \author Kim van der Riet - * - * Copyright (c) 2010 Red Hat, Inc. - * - * This file is part of the Qpid async store library msgstore.so. - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation; either - * version 2.1 of the License, or (at your option) any later version. - * - * This library is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 - * USA - * - * The GNU Lesser General Public License is available in the file COPYING. - */ - -#include "perf_test.hpp" - -#include <iostream> - -#include "jrnl/jexception.hpp" -#include "jrnl/slock.hpp" - -namespace mrg -{ -namespace test -{ - -perf_test::perf_test(const uint32_t num_msgs, - const uint32_t msg_size, - const char* msg_data, - mrg::journal::jcntl* const jp) : - _num_msgs(num_msgs), - _msg_size(msg_size), - _msg_data(msg_data), - _jp(jp) -{} - -perf_test::~perf_test() -{ - delete _jp; -} - -//*** MUST BE THREAD-SAFE **** -// This method will be called by multiple threads simultaneously -void -perf_test::operator()() -{ - // enqueue msgs - uint32_t msgs_enq = 0; - while (msgs_enq < _num_msgs) - { - mrg::journal::data_tok* dtokp = new mrg::journal::data_tok(); - mrg::journal::iores res = _jp->enqueue_data_record(_msg_data, _msg_size, _msg_size, dtokp); - switch (res) - { - case mrg::journal::RHM_IORES_SUCCESS: - msgs_enq++; -//std::cout << "ENQ: " << dtokp->status_str() << std::endl; - break; - case mrg::journal::RHM_IORES_ENQCAPTHRESH: - std::cout << "operator() RHM_IORES_ENQCAPTHRESH: " << dtokp->status_str() << std::endl; - delete dtokp; - ::usleep(10); - break; - default: - delete dtokp; - std::cerr << "enqueue_data_record FAILED with " << mrg::journal::iores_str(res) << std::endl; - } - - while (_ucl.size()) - { - { - std::unique_lock<std::mutex> l; - mrg::journal::data_tok* dtokp = _ucl.front(); - _ucl.pop_front(); - } - bool done = false; - while (!done) - { -//std::cout << " #"; - mrg::journal::iores res = _jp->dequeue_data_record(dtokp); - switch (res) - { - case mrg::journal::RHM_IORES_SUCCESS: -//std::cout << " -> OK" << std::endl; - done = true; - break; - case mrg::journal::RHM_IORES_BUSY: -//std::cout << " -> BUSY" << std::endl; - //_ucl.push_back(dtokp); - ::usleep(10); - break; - default: - std::cerr << "dequeue_data_record FAILED with " << mrg::journal::iores_str(res) << ": " << dtokp->status_str() << std::endl; - delete dtokp; - done = true; - } - } - - } - }; - - // flush - _jp->flush(false); - - // wait for all records to be written and dequeued - ::timespec timeout; - timeout.tv_sec = 2; - timeout.tv_nsec = 0; - while (_jp->get_wr_aio_evt_rem()) - { - _jp->get_wr_events(&timeout); - if (_jp->get_wr_aio_evt_rem()) ::usleep(10); - } - - // final flush - _jp->flush(true); -} - -//*** MUST BE THREAD-SAFE **** -// This method will be called by multiple threads simultaneously -void -perf_test::wr_aio_cb(std::vector<mrg::journal::data_tok*>& dtokl) -{ - mrg::journal::data_tok* dtokp; - int n = 0; -//std::cout << "=======------> wr_aio_cb(): " << &dtokl << " dtokl.size()=" << dtokl.size() << std::endl; - for (std::vector<mrg::journal::data_tok*>::const_iterator i=dtokl.begin(); i!=dtokl.end(); i++,n++) - { - dtokp = *i; -//std::cout << "DEQ " << n << ": " << dtokp << ": "; - switch (dtokp->wstate()) - { - case mrg::journal::data_tok::ENQ: - { - dtokp->set_dequeue_rid(dtokp->rid()); -//std::cout << dtokp->status_str() << std::endl; - bool done = false; - while (!done) - { -//std::cout << " *"; - mrg::journal::iores res = _jp->dequeue_data_record(dtokp); - switch (res) - { - case mrg::journal::RHM_IORES_SUCCESS: -//std::cout << " -> OK" << std::endl; - done = true; - break; - case mrg::journal::RHM_IORES_BUSY: -//std::cout << " -> BUSY" << std::endl; - { - std::unique_lock<std::mutex> l; - _ucl.push_back(dtokp); - } - done = true; - //::usleep(10); - break; - default: - std::cerr << "dequeue_data_record FAILED with " << mrg::journal::iores_str(res) << ": " << dtokp->status_str() << std::endl; - delete dtokp; - done = true; - } - } - } - break; - case mrg::journal::data_tok::DEQ: -//std::cout << "XXX: " << dtokp->status_str() << std::endl; - delete dtokp; - break; - default: ; - } - } -} - -//*** MUST BE THREAD-SAFE **** -// This method will be called by multiple threads simultaneously -void -perf_test::rd_aio_cb(std::vector<u_int16_t>& pil) -{} - -} // namespace test -} // namespace mrg Deleted: store/trunk/cpp/perf/perf_test.hpp =================================================================== --- store/trunk/cpp/perf/perf_test.hpp 2010-12-02 15:11:07 UTC (rev 4419) +++ store/trunk/cpp/perf/perf_test.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -1,68 +0,0 @@ -/** - * \file perf_test.hpp - * - * Qpid asynchronous store plugin library - * - * This file contains performance test code for the journal. - * - * \author Kim van der Riet - * - * Copyright (c) 2010 Red Hat, Inc. - * - * This file is part of the Qpid async store library msgstore.so. - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation; either - * version 2.1 of the License, or (at your option) any later version. - * - * This library is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 - * USA - * - * The GNU Lesser General Public License is available in the file COPYING. - */ - -#ifndef mrg_test_perf_test_hpp -#define mrg_test_perf_test_hpp - -#include <cstdint> -#include <deque> -#include <mutex> - -#include "jrnl/aio_callback.hpp" -#include "jrnl/data_tok.hpp" -#include "jrnl/jcntl.hpp" -#include "jrnl/smutex.hpp" - -namespace mrg -{ -namespace test -{ - - class perf_test: public mrg::journal::aio_callback - { - const uint32_t _num_msgs; - const uint32_t _msg_size; - const char* _msg_data; - mrg::journal::jcntl* const _jp; - std::deque<mrg::journal::data_tok*> _ucl; - std::mutex _ucl_mutex; - public: - perf_test(const uint32_t num_msgs, const uint32_t msg_size, const char* msg_data, mrg::journal::jcntl* const jp); - ~perf_test(); - void operator()(); - virtual void wr_aio_cb(std::vector<mrg::journal::data_tok*>& dtokl); - virtual void rd_aio_cb(std::vector<u_int16_t>& pil); - }; - -} // namespace test -} // namespace mrg - -#endif // mrg_test_perf_test_hpp Deleted: store/trunk/cpp/perf/perf_timer.cpp =================================================================== --- store/trunk/cpp/perf/perf_timer.cpp 2010-12-02 15:11:07 UTC (rev 4419) +++ store/trunk/cpp/perf/perf_timer.cpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -1,69 +0,0 @@ -/** - * \file timer.cpp - * - * Qpid asynchronous store plugin library - * - * This file contains performance test code for the journal. - * - * \author Kim van der Riet - * - * Copyright (c) 2010 Red Hat, Inc. - * - * This file is part of the Qpid async store library msgstore.so. - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation; either - * version 2.1 of the License, or (at your option) any later version. - * - * This library is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 - * USA - * - * The GNU Lesser General Public License is available in the file COPYING. - */ - -#include "perf_timer.hpp" -#include <iostream> - -namespace mrg -{ -namespace test -{ - perf_timer::perf_timer(const uint32_t num_msgs, const uint32_t msg_size, const uint16_t num_queues, const uint16_t num_threads_per_queue) : - _num_msgs(num_msgs), _msg_size(msg_size), _num_queues(num_queues), _num_threads_per_queue(num_threads_per_queue) - { - ::timespec ts; - ::clock_gettime(CLOCK_REALTIME, &ts); - _start_time = get_double_time(ts); - } - - perf_timer::~perf_timer() - { - ::timespec ts; - ::clock_gettime(CLOCK_REALTIME, &ts); - double time_taken = get_double_time(ts) - _start_time; - std::cout << " Msgs per thread: " << _num_msgs << std::endl; - std::cout << " Msg size: " << _msg_size << std::endl; - std::cout << " No. queues: " << _num_queues << std::endl; - std::cout << "No. threads/queue: " << _num_threads_per_queue << std::endl; - std::cout << " Time taken: " << time_taken << " sec" << std::endl; - std::cout << " Msg throughput: " << (_num_msgs * _num_queues * _num_threads_per_queue / time_taken / 1e3) << " kMsgs/sec" << std::endl; - std::cout << " " << (_num_msgs * _num_queues * _num_threads_per_queue * _msg_size / time_taken / 1e6) << " MB/sec" << std::endl; - } - - // static - double perf_timer::get_double_time(const ::timespec& ts) - { - return ts.tv_sec + (double(ts.tv_nsec) / 1e9); - } - - -} // namespace test -} // namespace mrg Deleted: store/trunk/cpp/perf/perf_timer.hpp =================================================================== --- store/trunk/cpp/perf/perf_timer.hpp 2010-12-02 15:11:07 UTC (rev 4419) +++ store/trunk/cpp/perf/perf_timer.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -1,59 +0,0 @@ -/** - * \file timer.hpp - * - * Qpid asynchronous store plugin library - * - * This file contains performance test code for the journal. - * - * \author Kim van der Riet - * - * Copyright (c) 2010 Red Hat, Inc. - * - * This file is part of the Qpid async store library msgstore.so. - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation; either - * version 2.1 of the License, or (at your option) any later version. - * - * This library is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 - * USA - * - * The GNU Lesser General Public License is available in the file COPYING. - */ - -#ifndef mrg_test_timer_hpp -#define mrg_test_timer_hpp - -#include <cstdint> -#include <ctime> - -namespace mrg -{ -namespace test -{ - - class perf_timer - { - uint32_t _num_msgs; - uint32_t _msg_size; - uint16_t _num_queues; - uint16_t _num_threads_per_queue; - double _start_time; - static double get_double_time(const timespec& tx); - public: - perf_timer(const uint32_t num_msgs, const uint32_t msg_size, const uint16_t num_queues, const uint16_t num_threads_per_queue); - ~perf_timer(); - }; - -} // namespace test -} // namespace mrg - -#endif // mrg_test_timer_hpp Deleted: store/trunk/cpp/perf/store_params.cpp =================================================================== --- store/trunk/cpp/perf/store_params.cpp 2010-12-02 15:11:07 UTC (rev 4419) +++ store/trunk/cpp/perf/store_params.cpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -1,93 +0,0 @@ -/** - * \file store_params.hpp - * - * Qpid asynchronous store plugin library - * - * This file contains performance test code for the journal. - * - * \author Kim van der Riet - * - * Copyright (c) 2010 Red Hat, Inc. - * - * This file is part of the Qpid async store library msgstore.so. - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation; either - * version 2.1 of the License, or (at your option) any later version. - * - * This library is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 - * USA - * - * The GNU Lesser General Public License is available in the file COPYING. - */ - -#include "store_params.hpp" - -#include <sstream> - -namespace mrg -{ -namespace test -{ - -// static -std::string store_params::_def_jrnl_dir("/tmp/store"); - -// static -std::string store_params::_def_jrnl_base_filename("JournalData"); - -store_params::store_params() : - _jrnl_dir(_def_jrnl_dir), - _jrnl_base_filename(_def_jrnl_base_filename), - _num_jfiles(8), - _jfsize_sblks(3072), - _auto_expand(false), - _ae_max_jfiles(0), - _wcache_num_pages(32), - _wcache_pgsize_sblks(128) -{} - -store_params::store_params(std::string& jrnl_dir, - std::string& jrnl_base_filename, - const u_int16_t num_jfiles, - const bool auto_expand, - const u_int16_t ae_max_jfiles, - const u_int32_t jfsize_sblks, - const u_int16_t wcache_num_pages, - const u_int32_t wcache_pgsize_sblks) : - _jrnl_dir(jrnl_dir), - _jrnl_base_filename(jrnl_base_filename), - _num_jfiles(num_jfiles), - _auto_expand(auto_expand), - _ae_max_jfiles(ae_max_jfiles), - _jfsize_sblks(jfsize_sblks), - _wcache_num_pages(wcache_num_pages), - _wcache_pgsize_sblks(wcache_pgsize_sblks) -{} - -std::string -store_params::to_string() -{ - std::ostringstream oss; - oss << "Store Parameters:" << std::endl; - oss << " jrnl_dir = " << _jrnl_dir << std::endl; - oss << " jrnl_base_filename = " << _jrnl_base_filename << std::endl; - oss << " num_jfiles = " << _num_jfiles << std::endl; - oss << " jfsize_sblks = " << _jfsize_sblks << std::endl; - oss << " auto_expand = " << _auto_expand << std::endl; - oss << " ae_max_jfiles = " << _ae_max_jfiles << std::endl; - oss << " wcache_num_pages = " << _wcache_num_pages << std::endl; - oss << " wcache_pgsize_sblks = " << _wcache_pgsize_sblks << std::endl; - return oss.str(); -} - -} // namespace test -} // namespace mrg Deleted: store/trunk/cpp/perf/store_params.hpp =================================================================== --- store/trunk/cpp/perf/store_params.hpp 2010-12-02 15:11:07 UTC (rev 4419) +++ store/trunk/cpp/perf/store_params.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -1,65 +0,0 @@ -/** - * \file store_params.hpp - * - * Qpid asynchronous store plugin library - * - * This file contains performance test code for the journal. - * - * \author Kim van der Riet - * - * Copyright (c) 2010 Red Hat, Inc. - * - * This file is part of the Qpid async store library msgstore.so. - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation; either - * version 2.1 of the License, or (at your option) any later version. - * - * This library is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 - * USA - * - * The GNU Lesser General Public License is available in the file COPYING. - */ - -#ifndef mrg_test_store_params_hpp -#define mrg_test_store_params_hpp - -#include <cstdint> -#include <string> - -namespace mrg -{ -namespace test -{ - - struct store_params - { - static std::string _def_jrnl_dir; - static std::string _def_jrnl_base_filename; - std::string& _jrnl_dir; - std::string& _jrnl_base_filename; - u_int16_t _num_jfiles; - u_int32_t _jfsize_sblks; - bool _auto_expand; - u_int16_t _ae_max_jfiles; - u_int16_t _wcache_num_pages; - u_int32_t _wcache_pgsize_sblks; - store_params(); - store_params(std::string& jrnl_dir, std::string& jrnl_base_filename, const u_int16_t num_jfiles, - const bool auto_expand, const u_int16_t ae_max_jfiles, const u_int32_t jfsize_sblks, - const u_int16_t wcache_num_pages, const u_int32_t wcache_pgsize_sblks); - std::string to_string(); - }; - -} // namespace test -} // namespace mrg - -#endif // mrg_test_store_params_hpp Deleted: store/trunk/cpp/perf/test_params.cpp =================================================================== --- store/trunk/cpp/perf/test_params.cpp 2010-12-02 15:11:07 UTC (rev 4419) +++ store/trunk/cpp/perf/test_params.cpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -1,69 +0,0 @@ -/** - * \file test_params.hpp - * - * Qpid asynchronous store plugin library - * - * This file contains performance test code for the journal. - * - * \author Kim van der Riet - * - * Copyright (c) 2010 Red Hat, Inc. - * - * This file is part of the Qpid async store library msgstore.so. - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation; either - * version 2.1 of the License, or (at your option) any later version. - * - * This library is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 - * USA - * - * The GNU Lesser General Public License is available in the file COPYING. - */ - -#include "test_params.hpp" - -#include <sstream> - -namespace mrg -{ -namespace test -{ - -test_params::test_params(): - _num_msgs(100), - _msg_size(10), - _num_queues(1), - _num_threads_per_queue(1) -{} - -test_params::test_params(const uint32_t num_msgs, const uint32_t msg_size, const uint16_t num_queues, - const uint16_t num_threads_per_queue) : - _num_msgs(num_msgs), - _msg_size(msg_size), - _num_queues(num_queues), - _num_threads_per_queue(num_threads_per_queue) -{} - -std::string -test_params::to_string() -{ - std::ostringstream oss; - oss << "Test Parameters:" << std::endl; - oss << " num_msgs = " << _num_msgs << std::endl; - oss << " msg_size = " << _msg_size << std::endl; - oss << " num_queues = " << _num_queues << std::endl; - oss << " num_threads_per_queue = " << _num_threads_per_queue << std::endl; - return oss.str(); -} - -} // namespace test -} // namespace mrg Deleted: store/trunk/cpp/perf/test_params.hpp =================================================================== --- store/trunk/cpp/perf/test_params.hpp 2010-12-02 15:11:07 UTC (rev 4419) +++ store/trunk/cpp/perf/test_params.hpp 2010-12-02 20:48:07 UTC (rev 4420) @@ -1,58 +0,0 @@ -/** - * \file test_params.hpp - * - * Qpid asynchronous store plugin library - * - * This file contains performance test code for the journal. - * - * \author Kim van der Riet - * - * Copyright (c) 2010 Red Hat, Inc. - * - * This file is part of the Qpid async store library msgstore.so. - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation; either - * version 2.1 of the License, or (at your option) any later version. - * - * This library is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 - * USA - * - * The GNU Lesser General Public License is available in the file COPYING. - */ - -#ifndef mrg_test_test_params_hpp -#define mrg_test_test_params_hpp - -#include <cstdint> -#include <string> - -namespace mrg -{ -namespace test -{ - - struct test_params - { - uint32_t _num_msgs; - uint32_t _msg_size; - uint16_t _num_queues; - uint16_t _num_threads_per_queue; - test_params(); - test_params(const uint32_t num_msgs, const uint32_t msg_size, const uint16_t num_queues, - const uint16_t num_threads_per_queue); - std::string to_string(); - }; - -} // namespace test -} // namespace mrg - -#endif // mrg_test_test_params_hpp