cpp/src/qpid/legacystore/JournalImpl.h - qpid - Git at Google

 /*
  *
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  *
  */

 #ifndef QPID_LEGACYSTORE_JOURNALIMPL_H
 #define QPID_LEGACYSTORE_JOURNALIMPL_H

 #include <set>
 #include "qpid/legacystore/jrnl/enums.h"
 #include "qpid/legacystore/jrnl/jcntl.h"
 #include "qpid/legacystore/DataTokenImpl.h"
 #include "qpid/legacystore/PreparedTransaction.h"
 #include "qpid/broker/PersistableQueue.h"
 #include "qpid/sys/Timer.h"
 #include "qpid/sys/Time.h"
 #include <boost/ptr_container/ptr_list.hpp>
 #include <boost/intrusive_ptr.hpp>
 #include "qpid/management/Manageable.h"
 #include "qmf/org/apache/qpid/legacystore/Journal.h"

 namespace qpid { namespace sys {
 class Timer;
 }}

 namespace mrg {
 namespace msgstore {

 class JournalImpl;

 class InactivityFireEvent : public qpid::sys::TimerTask
 {
     JournalImpl* _parent;
     qpid::sys::Mutex _ife_lock;

   public:
     InactivityFireEvent(JournalImpl* p, const qpid::sys::Duration timeout);
     virtual ~InactivityFireEvent() {}
     void fire();
     inline void cancel() { qpid::sys::Mutex::ScopedLock sl(_ife_lock); _parent = 0; }
 };

 class GetEventsFireEvent : public qpid::sys::TimerTask
 {
     JournalImpl* _parent;
     qpid::sys::Mutex _gefe_lock;

   public:
     GetEventsFireEvent(JournalImpl* p, const qpid::sys::Duration timeout);
     virtual ~GetEventsFireEvent() {}
     void fire();
     inline void cancel() { qpid::sys::Mutex::ScopedLock sl(_gefe_lock); _parent = 0; }
 };

 class JournalImpl : public qpid::broker::ExternalQueueStore, public mrg::journal::jcntl, public mrg::journal::aio_callback
 {
   public:
     typedef boost::function<void (JournalImpl&)> DeleteCallback;

   private:
 //    static qpid::sys::Mutex _static_lock;
 //    static u_int32_t cnt;

     qpid::sys::Timer& timer;
     bool getEventsTimerSetFlag;
     boost::intrusive_ptr<qpid::sys::TimerTask> getEventsFireEventsPtr;
     qpid::sys::Mutex _getf_lock;
     qpid::sys::Mutex _read_lock;

     u_int64_t lastReadRid; // rid of last read msg for loadMsgContent() - detects out-of-order read requests
     std::vector<u_int64_t> oooRidList; // list of out-of-order rids (greater than current rid) encountered during read sequence

     bool writeActivityFlag;
     bool flushTriggeredFlag;
     boost::intrusive_ptr<qpid::sys::TimerTask> inactivityFireEventPtr;

     // temp local vars for loadMsgContent below
     void* _xidp;
     void* _datap;
     size_t _dlen;
     mrg::journal::data_tok _dtok;
     bool _external;

     qpid::management::ManagementAgent* _agent;
     qmf::org::apache::qpid::legacystore::Journal::shared_ptr _mgmtObject;
     DeleteCallback deleteCallback;

   public:

     JournalImpl(qpid::sys::Timer& timer,
                 const std::string& journalId,
                 const std::string& journalDirectory,
                 const std::string& journalBaseFilename,
                 const qpid::sys::Duration getEventsTimeout,
                 const qpid::sys::Duration flushTimeout,
                 qpid::management::ManagementAgent* agent,
                 DeleteCallback deleteCallback=DeleteCallback() );

     virtual ~JournalImpl();

     void initManagement(qpid::management::ManagementAgent* agent);

     void initialize(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,
                     mrg::journal::aio_callback* const cbp);

     inline void initialize(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) {
         initialize(num_jfiles, auto_expand, ae_max_jfiles, jfsize_sblks, wcache_num_pages, wcache_pgsize_sblks,
                    this);
     }

     void recover(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,
                  mrg::journal::aio_callback* const cbp,
                  boost::ptr_list<msgstore::PreparedTransaction>* prep_tx_list_ptr,
                  u_int64_t& highest_rid,
                  u_int64_t queue_id);

     inline void recover(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,
                         boost::ptr_list<msgstore::PreparedTransaction>* prep_tx_list_ptr,
                         u_int64_t& highest_rid,
                         u_int64_t queue_id) {
         recover(num_jfiles, auto_expand, ae_max_jfiles, jfsize_sblks, wcache_num_pages, wcache_pgsize_sblks,
                 this, prep_tx_list_ptr, highest_rid, queue_id);
     }

     void recover_complete();

     // Temporary fn to read and save last msg read from journal so it can be assigned
     // in chunks. To be replaced when coding to do this direct from the journal is ready.
     // Returns true if the record is extern, false if local.
     bool loadMsgContent(u_int64_t rid, std::string& data, size_t length, size_t offset = 0);

     // Overrides for write inactivity timer
     void enqueue_data_record(const void* const data_buff, const size_t tot_data_len,
                              const size_t this_data_len, mrg::journal::data_tok* dtokp,
                              const bool transient = false);

     void enqueue_extern_data_record(const size_t tot_data_len, mrg::journal::data_tok* dtokp,
                                     const bool transient = false);

     void enqueue_txn_data_record(const void* const data_buff, const size_t tot_data_len,
                                  const size_t this_data_len, mrg::journal::data_tok* dtokp, const std::string& xid,
                                  const bool transient = false);

     void enqueue_extern_txn_data_record(const size_t tot_data_len, mrg::journal::data_tok* dtokp,
                                         const std::string& xid, const bool transient = false);

     void dequeue_data_record(mrg::journal::data_tok* const dtokp, const bool txn_coml_commit = false);

     void dequeue_txn_data_record(mrg::journal::data_tok* const dtokp, const std::string& xid, const bool txn_coml_commit = false);

     void txn_abort(mrg::journal::data_tok* const dtokp, const std::string& xid);

     void txn_commit(mrg::journal::data_tok* const dtokp, const std::string& xid);

     void stop(bool block_till_aio_cmpl = false);

     // Logging
     void log(mrg::journal::log_level level, const std::string& log_stmt) const;
     void log(mrg::journal::log_level level, const char* const log_stmt) const;

     // Overrides for get_events timer
     mrg::journal::iores flush(const bool block_till_aio_cmpl = false);

     // TimerTask callback
     void getEventsFire();
     void flushFire();

     // AIO callbacks
     virtual void wr_aio_cb(std::vector<mrg::journal::data_tok*>& dtokl);
     virtual void rd_aio_cb(std::vector<u_int16_t>& pil);

     qpid::management::ManagementObject::shared_ptr GetManagementObject (void) const
     { return _mgmtObject; }

     qpid::management::Manageable::status_t ManagementMethod (uint32_t,
                                                              qpid::management::Args&,
                                                              std::string&);

     void resetDeleteCallback() { deleteCallback = DeleteCallback(); }

   private:
     void free_read_buffers();

     inline void setGetEventTimer()
     {
         getEventsFireEventsPtr->setupNextFire();
         timer.add(getEventsFireEventsPtr);
         getEventsTimerSetFlag = true;
     }
     void handleIoResult(const mrg::journal::iores r);

     // Management instrumentation callbacks overridden from jcntl
     inline void instr_incr_outstanding_aio_cnt() {
       if (_mgmtObject.get() != 0) _mgmtObject->inc_outstandingAIOs();
     }
     inline void instr_decr_outstanding_aio_cnt() {
       if (_mgmtObject.get() != 0) _mgmtObject->dec_outstandingAIOs();
     }

 }; // class JournalImpl

 class TplJournalImpl : public JournalImpl
 {
   public:
     TplJournalImpl(qpid::sys::Timer& timer,
                    const std::string& journalId,
                    const std::string& journalDirectory,
                    const std::string& journalBaseFilename,
                    const qpid::sys::Duration getEventsTimeout,
                    const qpid::sys::Duration flushTimeout,
                    qpid::management::ManagementAgent* agent) :
         JournalImpl(timer, journalId, journalDirectory, journalBaseFilename, getEventsTimeout, flushTimeout, agent)
     {}

     virtual ~TplJournalImpl() {}

     // Special version of read_data_record that ignores transactions - needed when reading the TPL
     inline mrg::journal::iores read_data_record(void** const datapp, std::size_t& dsize,
                                                 void** const xidpp, std::size_t& xidsize, bool& transient, bool& external,
                                                 mrg::journal::data_tok* const dtokp) {
         return JournalImpl::read_data_record(datapp, dsize, xidpp, xidsize, transient, external, dtokp, true);
     }
     inline void read_reset() { _rmgr.invalidate(); }
 }; // class TplJournalImpl

 } // namespace msgstore
 } // namespace mrg

 #endif // ifndef QPID_LEGACYSTORE_JOURNALIMPL_H
	/*
	*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*
	*/

	#ifndef QPID_LEGACYSTORE_JOURNALIMPL_H
	#define QPID_LEGACYSTORE_JOURNALIMPL_H

	#include <set>
	#include "qpid/legacystore/jrnl/enums.h"
	#include "qpid/legacystore/jrnl/jcntl.h"
	#include "qpid/legacystore/DataTokenImpl.h"
	#include "qpid/legacystore/PreparedTransaction.h"
	#include "qpid/broker/PersistableQueue.h"
	#include "qpid/sys/Timer.h"
	#include "qpid/sys/Time.h"
	#include <boost/ptr_container/ptr_list.hpp>
	#include <boost/intrusive_ptr.hpp>
	#include "qpid/management/Manageable.h"
	#include "qmf/org/apache/qpid/legacystore/Journal.h"

	namespace qpid { namespace sys {
	class Timer;
	}}

	namespace mrg {
	namespace msgstore {

	class JournalImpl;

	class InactivityFireEvent : public qpid::sys::TimerTask
	{
	JournalImpl* _parent;
	qpid::sys::Mutex _ife_lock;

	public:
	InactivityFireEvent(JournalImpl* p, const qpid::sys::Duration timeout);
	virtual ~InactivityFireEvent() {}
	void fire();
	inline void cancel() { qpid::sys::Mutex::ScopedLock sl(_ife_lock); _parent = 0; }
	};

	class GetEventsFireEvent : public qpid::sys::TimerTask
	{
	JournalImpl* _parent;
	qpid::sys::Mutex _gefe_lock;

	public:
	GetEventsFireEvent(JournalImpl* p, const qpid::sys::Duration timeout);
	virtual ~GetEventsFireEvent() {}
	void fire();
	inline void cancel() { qpid::sys::Mutex::ScopedLock sl(_gefe_lock); _parent = 0; }
	};

	class JournalImpl : public qpid::broker::ExternalQueueStore, public mrg::journal::jcntl, public mrg::journal::aio_callback
	{
	public:
	typedef boost::function<void (JournalImpl&)> DeleteCallback;

	private:
	// static qpid::sys::Mutex _static_lock;
	// static u_int32_t cnt;

	qpid::sys::Timer& timer;
	bool getEventsTimerSetFlag;
	boost::intrusive_ptr<qpid::sys::TimerTask> getEventsFireEventsPtr;
	qpid::sys::Mutex _getf_lock;
	qpid::sys::Mutex _read_lock;

	u_int64_t lastReadRid; // rid of last read msg for loadMsgContent() - detects out-of-order read requests
	std::vector<u_int64_t> oooRidList; // list of out-of-order rids (greater than current rid) encountered during read sequence

	bool writeActivityFlag;
	bool flushTriggeredFlag;
	boost::intrusive_ptr<qpid::sys::TimerTask> inactivityFireEventPtr;

	// temp local vars for loadMsgContent below
	void* _xidp;
	void* _datap;
	size_t _dlen;
	mrg::journal::data_tok _dtok;
	bool _external;

	qpid::management::ManagementAgent* _agent;
	qmf::org::apache::qpid::legacystore::Journal::shared_ptr _mgmtObject;
	DeleteCallback deleteCallback;

	public:

	JournalImpl(qpid::sys::Timer& timer,
	const std::string& journalId,
	const std::string& journalDirectory,
	const std::string& journalBaseFilename,
	const qpid::sys::Duration getEventsTimeout,
	const qpid::sys::Duration flushTimeout,
	qpid::management::ManagementAgent* agent,
	DeleteCallback deleteCallback=DeleteCallback() );

	virtual ~JournalImpl();

	void initManagement(qpid::management::ManagementAgent* agent);

	void initialize(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,
	mrg::journal::aio_callback* const cbp);

	inline void initialize(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) {
	initialize(num_jfiles, auto_expand, ae_max_jfiles, jfsize_sblks, wcache_num_pages, wcache_pgsize_sblks,
	this);
	}

	void recover(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,
	mrg::journal::aio_callback* const cbp,
	boost::ptr_list<msgstore::PreparedTransaction>* prep_tx_list_ptr,
	u_int64_t& highest_rid,
	u_int64_t queue_id);

	inline void recover(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,
	boost::ptr_list<msgstore::PreparedTransaction>* prep_tx_list_ptr,
	u_int64_t& highest_rid,
	u_int64_t queue_id) {
	recover(num_jfiles, auto_expand, ae_max_jfiles, jfsize_sblks, wcache_num_pages, wcache_pgsize_sblks,
	this, prep_tx_list_ptr, highest_rid, queue_id);
	}

	void recover_complete();

	// Temporary fn to read and save last msg read from journal so it can be assigned
	// in chunks. To be replaced when coding to do this direct from the journal is ready.
	// Returns true if the record is extern, false if local.
	bool loadMsgContent(u_int64_t rid, std::string& data, size_t length, size_t offset = 0);

	// Overrides for write inactivity timer
	void enqueue_data_record(const void* const data_buff, const size_t tot_data_len,
	const size_t this_data_len, mrg::journal::data_tok* dtokp,
	const bool transient = false);

	void enqueue_extern_data_record(const size_t tot_data_len, mrg::journal::data_tok* dtokp,
	const bool transient = false);

	void enqueue_txn_data_record(const void* const data_buff, const size_t tot_data_len,
	const size_t this_data_len, mrg::journal::data_tok* dtokp, const std::string& xid,
	const bool transient = false);

	void enqueue_extern_txn_data_record(const size_t tot_data_len, mrg::journal::data_tok* dtokp,
	const std::string& xid, const bool transient = false);

	void dequeue_data_record(mrg::journal::data_tok* const dtokp, const bool txn_coml_commit = false);

	void dequeue_txn_data_record(mrg::journal::data_tok* const dtokp, const std::string& xid, const bool txn_coml_commit = false);

	void txn_abort(mrg::journal::data_tok* const dtokp, const std::string& xid);

	void txn_commit(mrg::journal::data_tok* const dtokp, const std::string& xid);

	void stop(bool block_till_aio_cmpl = false);

	// Logging
	void log(mrg::journal::log_level level, const std::string& log_stmt) const;
	void log(mrg::journal::log_level level, const char* const log_stmt) const;

	// Overrides for get_events timer
	mrg::journal::iores flush(const bool block_till_aio_cmpl = false);

	// TimerTask callback
	void getEventsFire();
	void flushFire();

	// AIO callbacks
	virtual void wr_aio_cb(std::vector<mrg::journal::data_tok*>& dtokl);
	virtual void rd_aio_cb(std::vector<u_int16_t>& pil);

	qpid::management::ManagementObject::shared_ptr GetManagementObject (void) const
	{ return _mgmtObject; }

	qpid::management::Manageable::status_t ManagementMethod (uint32_t,
	qpid::management::Args&,
	std::string&);

	void resetDeleteCallback() { deleteCallback = DeleteCallback(); }

	private:
	void free_read_buffers();

	inline void setGetEventTimer()
	{
	getEventsFireEventsPtr->setupNextFire();
	timer.add(getEventsFireEventsPtr);
	getEventsTimerSetFlag = true;
	}
	void handleIoResult(const mrg::journal::iores r);

	// Management instrumentation callbacks overridden from jcntl
	inline void instr_incr_outstanding_aio_cnt() {
	if (_mgmtObject.get() != 0) _mgmtObject->inc_outstandingAIOs();
	}
	inline void instr_decr_outstanding_aio_cnt() {
	if (_mgmtObject.get() != 0) _mgmtObject->dec_outstandingAIOs();
	}

	}; // class JournalImpl

	class TplJournalImpl : public JournalImpl
	{
	public:
	TplJournalImpl(qpid::sys::Timer& timer,
	const std::string& journalId,
	const std::string& journalDirectory,
	const std::string& journalBaseFilename,
	const qpid::sys::Duration getEventsTimeout,
	const qpid::sys::Duration flushTimeout,
	qpid::management::ManagementAgent* agent) :
	JournalImpl(timer, journalId, journalDirectory, journalBaseFilename, getEventsTimeout, flushTimeout, agent)
	{}

	virtual ~TplJournalImpl() {}

	// Special version of read_data_record that ignores transactions - needed when reading the TPL
	inline mrg::journal::iores read_data_record(void** const datapp, std::size_t& dsize,
	void** const xidpp, std::size_t& xidsize, bool& transient, bool& external,
	mrg::journal::data_tok* const dtokp) {
	return JournalImpl::read_data_record(datapp, dsize, xidpp, xidsize, transient, external, dtokp, true);
	}
	inline void read_reset() { _rmgr.invalidate(); }
	}; // class TplJournalImpl

	} // namespace msgstore
	} // namespace mrg

	#endif // ifndef QPID_LEGACYSTORE_JOURNALIMPL_H