OpenDDS::DCPS::DataDurabilityCache Class Reference

Underlying data cache for both OpenDDS TRANSIENT and PERSISTENT DURABILITY implementations.. More...

#include <DataDurabilityCache.h>

Collaboration diagram for OpenDDS::DCPS::DataDurabilityCache:

Classes
class	key_type
	Key type for underlying maps. More...
class	sample_data_type
	Sample list data type for all samples. More...

Public Types
typedef DurabilityArray< DurabilityQueue< sample_data_type > * >	sample_list_type
typedef ACE_Hash_Map_With_Allocator< key_type, sample_list_type * >	sample_map_type

Public Member Functions
typedef	OPENDDS_LIST (long) timer_id_list_type
	DataDurabilityCache (DDS::DurabilityQosPolicyKind kind)
	DataDurabilityCache (DDS::DurabilityQosPolicyKind kind, ACE_CString &data_dir)
	~DataDurabilityCache ()
bool	insert (DDS::DomainId_t domain_id, char const *topic_name, char const *type_name, SendStateDataSampleList &the_data, DDS::DurabilityServiceQosPolicy const &qos)
bool	get_data (DDS::DomainId_t domain_id, char const *topic_name, char const *type_name, DataWriterImpl *data_writer, ACE_Allocator *mb_allocator, ACE_Allocator *db_allocator, DDS::LifespanQosPolicy const &)

Private Member Functions
	DataDurabilityCache (DataDurabilityCache const &)
DataDurabilityCache &	operator= (DataDurabilityCache const &)
void	init ()

Private Attributes
unique_ptr< ACE_Allocator > const	allocator_
	Allocator used to allocate memory for sample map and lists. More...
DDS::DurabilityQosPolicyKind	kind_
ACE_CString	data_dir_
sample_map_type *	samples_
	Map of all data samples. More...
timer_id_list_type	cleanup_timer_ids_
	Timer ID list. More...
ACE_SYNCH_MUTEX	lock_
	Lock for synchronized access to the underlying map. More...
ACE_Reactor_Timer_Interface *	reactor_
	Reactor with which cleanup timers will be registered. More...

Detailed Description

Underlying data cache for both OpenDDS TRANSIENT and PERSISTENT DURABILITY implementations..

This class implements a cache that outlives DataWriters.

Definition at line 69 of file DataDurabilityCache.h.

Member Typedef Documentation

sample_list_type

typedef DurabilityArray< DurabilityQueue<sample_data_type> *> OpenDDS::DCPS::DataDurabilityCache::sample_list_type

Definition at line 184 of file DataDurabilityCache.h.

sample_map_type

typedef ACE_Hash_Map_With_Allocator<key_type, sample_list_type *> OpenDDS::DCPS::DataDurabilityCache::sample_map_type

Definition at line 187 of file DataDurabilityCache.h.

Constructor & Destructor Documentation

DataDurabilityCache() [1/3]

OpenDDS::DCPS::DataDurabilityCache::DataDurabilityCache

(

DDS::DurabilityQosPolicyKind

kind

)

Definition at line 287 of file DataDurabilityCache.cpp.

References init().

  : allocator_(new ACE_New_Allocator)
  , kind_(kind)
  , samples_(0)
  , cleanup_timer_ids_()
  , lock_()
  , reactor_(0)
{
  init();
}

DataDurabilityCache() [2/3]

OpenDDS::DCPS::DataDurabilityCache::DataDurabilityCache	(	DDS::DurabilityQosPolicyKind	kind,
		ACE_CString &	data_dir
	)

Definition at line 299 of file DataDurabilityCache.cpp.

References init().

  : allocator_(new ACE_New_Allocator)
  , kind_(kind)
  , data_dir_(data_dir)
  , samples_(0)
  , cleanup_timer_ids_()
  , lock_()
  , reactor_(0)
{
  init();
}

~DataDurabilityCache()

OpenDDS::DCPS::DataDurabilityCache::~DataDurabilityCache

(

)

Definition at line 427 of file DataDurabilityCache.cpp.

References ACE_DES_FREE, allocator_, ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, ACE_Hash< EXT_ID >, ACE_Equal_To< EXT_ID >, ACE_Null_Mutex >::begin(), ACE_Reactor_Timer_Interface::cancel_timer(), cleanup_timer_ids_, ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, ACE_Hash< EXT_ID >, ACE_Equal_To< EXT_ID >, ACE_Null_Mutex >::end(), ACE_Allocator::free(), OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), OPENDDS_MAP_TYPE, reactor_, samples_, and ACE_Array_Base< T >::size().

{
  // Cancel timers that haven't expired yet.
  timer_id_list_type::const_iterator const end(
    this->cleanup_timer_ids_.end());

  for (timer_id_list_type::const_iterator i(
         this->cleanup_timer_ids_.begin());
       i != end;
       ++i) {
    (void) this->reactor_->cancel_timer(*i);
  }

  // Clean up memory that isn't automatically managed.
  if (this->allocator_.get() != 0) {
    sample_map_type::iterator const map_end = this->samples_->end();

    for (sample_map_type::iterator s = this->samples_->begin();
         s != map_end;
         ++s) {
      sample_list_type * const list = (*s).int_id_;

      size_t const len = list->size();

      for (size_t l = 0; l != len; ++l) {
        ACE_DES_FREE((*list)[l],
                     this->allocator_->free,
                     DurabilityQueue<sample_data_type>);
      }

      ACE_DES_FREE(list,
                   this->allocator_->free,
                   DurabilityArray<DurabilityQueue<sample_data_type> *>);
    }

    // Yes, this looks strange but please leave it in place.  The third param
    // to ACE_DES_FREE must be the actual class name since it's used in an
    // explicit desturctor call (~T()).  Typedefs are not allowed here.  This
    // is why the two ACE_DES_FREE's above are not the typedefs.  Below we use
    // a macro to hide the internal comma from ACE_DES_FREE's macro expansion.
#define OPENDDS_MAP_TYPE ACE_Hash_Map_With_Allocator<key_type, sample_list_type *>
    ACE_DES_FREE(this->samples_, this->allocator_->free, OPENDDS_MAP_TYPE);
#undef OPENDDS_MAP_TYPE
  }
}

DataDurabilityCache() [3/3]

OpenDDS::DCPS::DataDurabilityCache::DataDurabilityCache ( DataDurabilityCache const &  )

private

Member Function Documentation

get_data()

bool OpenDDS::DCPS::DataDurabilityCache::get_data	(	DDS::DomainId_t	domain_id,
		char const *	topic_name,
		char const *	type_name,
		DataWriterImpl *	data_writer,
		ACE_Allocator *	mb_allocator,
		ACE_Allocator *	db_allocator,
		DDS::LifespanQosPolicy const &
	)

Write cached data corresponding to given domain, topic and type to DataWriter.

Todo:

Is this going to cause problems for users that set a finite DDS::ResourceLimitsQosPolicy::max_instances value when OpenDDS supports that value?

Todo:

If we don't empty the queue, we'll end up with duplicate data since the data retrieved from the cache will be reinserted.

Definition at line 718 of file DataDurabilityCache.cpp.

References ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY, ACE_ERROR, ACE_GUARD_RETURN, ACE_NEW_MALLOC_RETURN, ACE_SYNCH_MUTEX, ACE_TEXT(), ACE_Array_Base< T >::allocator_, allocator_, OpenDDS::DCPS::DCPS_debug_level, ACE_Hash_Map_With_Allocator< class, class >::find(), OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), OpenDDS::DCPS::DataWriterImpl::get_db_lock(), OpenDDS::DCPS::DataDurabilityCache::sample_data_type::get_sample(), DDS::HANDLE_NIL, LM_ERROR, lock_, ACE_Allocator::malloc(), ACE_Time_Value::max_time, ACE_Message_Block::MB_DATA, ACE_OS::memcpy(), OpenDDS::DCPS::move(), OpenDDS::DCPS::DataWriterImpl::register_instance_from_durable_data(), DDS::RETCODE_OK, samples_, ACE_Array_Base< T >::size(), ACE_Message_Block::wr_ptr(), OpenDDS::DCPS::DataWriterImpl::write(), and ACE_Time_Value::zero.

{
  key_type const key(domain_id,
                     topic_name,
                     type_name,
                     this->allocator_.get());

  ACE_GUARD_RETURN(ACE_SYNCH_MUTEX, guard, this->lock_, false);

  sample_list_type * p_sample_list = 0;

  if (this->samples_->find(key,
                           p_sample_list,
                           this->allocator_.get()) == -1)
    return true;  // No durable data for this domain/topic/type.

  else if (p_sample_list == 0)
    return false; // Should never happen.

  sample_list_type & sample_list = *p_sample_list;

  // We will register an instance, and then write all of the cached
  // data to the DataWriter using that instance.

  sample_data_type * registration_data = 0;

  if (sample_list[0]->get(registration_data, 0) == -1)
    return false;

  char const * marshaled_sample = 0;
  size_t marshaled_sample_length = 0;
  DDS::Time_t registration_timestamp;

  registration_data->get_sample(marshaled_sample,
                                marshaled_sample_length,
                                registration_timestamp);

  // Don't use the cached allocator for the registered sample message
  // block.
  Message_Block_Ptr registration_sample(
    new ACE_Message_Block(marshaled_sample_length,
                          ACE_Message_Block::MB_DATA,
                          0, //cont
                          0, //data
                          0, //alloc_strategy
                          data_writer->get_db_lock()));

  ACE_OS::memcpy(registration_sample->wr_ptr(),
                 marshaled_sample,
                 marshaled_sample_length);

  registration_sample->wr_ptr(marshaled_sample_length);

  DDS::InstanceHandle_t handle = DDS::HANDLE_NIL;

  /**
   * @todo Is this going to cause problems for users that set a finite
   *       DDS::ResourceLimitsQosPolicy::max_instances value when
   *       OpenDDS supports that value?
   */
  DDS::ReturnCode_t ret =
    data_writer->register_instance_from_durable_data(handle,
                                     move(registration_sample),
                                     registration_timestamp);

  if (ret != DDS::RETCODE_OK)
    return false;

  typedef DurabilityQueue<sample_data_type> data_queue_type;
  size_t const len = sample_list.size();

  for (size_t i = 0; i != len; ++i) {
    data_queue_type * const q = sample_list[i];

    for (data_queue_type::ITERATOR j = q->begin();
         !j.done();
         j.advance()) {
      sample_data_type * data = 0;

      if (j.next(data) == 0)
        return false;  // Should never happen.

      char const * sample = 0;  // Sample does not include header.
      size_t sample_length = 0;
      DDS::Time_t source_timestamp;

      data->get_sample(sample, sample_length, source_timestamp);

      ACE_Message_Block * tmp_mb = 0;
      ACE_NEW_MALLOC_RETURN(tmp_mb,
                            static_cast<ACE_Message_Block*>(
                              mb_allocator->malloc(
                                sizeof(ACE_Message_Block))),
                            ACE_Message_Block(
                              sample_length,
                              ACE_Message_Block::MB_DATA,
                              0, // cont
                              0, // data
                              0, // allocator_strategy
                              data_writer->get_db_lock(), // data block locking_strategy
                              ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY,
                              ACE_Time_Value::zero,
                              ACE_Time_Value::max_time,
                              db_allocator,
                              mb_allocator),
                            false);
      Message_Block_Ptr mb(tmp_mb);

      ACE_OS::memcpy(mb->wr_ptr(),
                     sample,
                     sample_length);
      mb->wr_ptr(sample_length);

      const DDS::ReturnCode_t ret = data_writer->write(move(mb),
                                                       handle,
                                                       source_timestamp,
                                                       0 /* no content filtering */,
                                                       0 /* no pointer to data */);

      if (ret != DDS::RETCODE_OK) {
        return false;
      }
    }

    // Data successfully written.  Empty the queue/list.
    /**
     * @todo If we don't empty the queue, we'll end up with duplicate
     *       data since the data retrieved from the cache will be
     *       reinserted.
     */
    q->reset();

    try {
      cleanup_directory(q->fs_path_, this->data_dir_);

    } catch (const std::exception& ex) {
      if (DCPS_debug_level > 0) {
        ACE_ERROR((LM_ERROR,
                   ACE_TEXT("(%P|%t) DataDurabilityCache::get_data ")
                   ACE_TEXT("couldn't remove directory for PERSISTENT ")
                   ACE_TEXT("data: %C\n"), ex.what()));
      }
    }
  }
  return true;
}

init()

void OpenDDS::DCPS::DataDurabilityCache::init ( void  )

private

Definition at line 313 of file DataDurabilityCache.cpp.

References ACE_ERROR, ACE_NEW_MALLOC, ACE_TEXT(), allocator_, ACE_OS::atoi(), ACE_Hash_Map_With_Allocator< class, class >::bind(), ACE_String_Base< char >::c_str(), ACE_Message_Block::cont(), data_dir_, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), kind_, LM_ERROR, ACE_Allocator::malloc(), DDS::Time_t::nanosec, OPENDDS_STRING, OpenDDS::DCPS::OPENDDS_VECTOR(), DDS::PERSISTENT_DURABILITY_QOS, reactor_, ACE_Message_Block::release(), samples_, DDS::Time_t::sec, ACE_Array_Base< T >::size(), ACE_Message_Block::space(), TheServiceParticipant, timestamp(), and ACE_Message_Block::wr_ptr().

Referenced by DataDurabilityCache().

{
  ACE_Allocator * const allocator = this->allocator_.get();
  ACE_NEW_MALLOC(
    this->samples_,
    static_cast<sample_map_type *>(
      allocator->malloc(sizeof(sample_map_type))),
    sample_map_type(allocator));

  typedef DurabilityQueue<sample_data_type> data_queue_type;

  if (this->kind_ == DDS::PERSISTENT_DURABILITY_QOS) {
    // Read data from the filesystem and create the in-memory data structures
    // as if we had called insert() once for each "datawriter" directory.
    using OpenDDS::FileSystemStorage::Directory;
    using OpenDDS::FileSystemStorage::File;
    Directory::Ptr root_dir = Directory::create(this->data_dir_.c_str());
    OPENDDS_VECTOR(OPENDDS_STRING) path(4);  // domain, topic, type, datawriter

    for (Directory::DirectoryIterator domain = root_dir->begin_dirs(),
         domain_end = root_dir->end_dirs(); domain != domain_end; ++domain) {
      path[0] = domain->name();
      DDS::DomainId_t domain_id;
      {
        domain_id = ACE_OS::atoi(path[0].c_str());
      }

      for (Directory::DirectoryIterator topic = domain->begin_dirs(),
           topic_end = domain->end_dirs(); topic != topic_end; ++topic) {
        path[1] = topic->name();

        for (Directory::DirectoryIterator type = topic->begin_dirs(),
             type_end = topic->end_dirs(); type != type_end; ++type) {
          path[2] = type->name();

          key_type key(domain_id, path[1].c_str(), path[2].c_str(),
                       allocator);
          sample_list_type * sample_list = 0;
          ACE_NEW_MALLOC(sample_list,
                         static_cast<sample_list_type *>(
                           allocator->malloc(sizeof(sample_list_type))),
                         sample_list_type(0, static_cast<data_queue_type *>(0),
                                          allocator));
          this->samples_->bind(key, sample_list, allocator);

          for (Directory::DirectoryIterator dw = type->begin_dirs(),
               dw_end = type->end_dirs(); dw != dw_end; ++dw) {
            path[3] = dw->name();

            size_t old_len = sample_list->size();
            sample_list->size(old_len + 1);
            data_queue_type *& slot = (*sample_list)[old_len];

            // This variable is called "samples" in the insert() method be
            // we already have a "samples_" which is the overall data structure.
            data_queue_type * sample_queue = 0;
            ACE_NEW_MALLOC(sample_queue,
                           static_cast<data_queue_type *>(
                             allocator->malloc(sizeof(data_queue_type))),
                           data_queue_type(allocator));

            slot = sample_queue;
            sample_queue->fs_path_ = path;

            for (Directory::FileIterator file = dw->begin_files(),
                 file_end = dw->end_files(); file != file_end; ++file) {
              std::ifstream is;

              if (!file->read(is)) {
                if (DCPS_debug_level) {
                  ACE_ERROR((LM_ERROR,
                             ACE_TEXT("(%P|%t) DataDurabilityCache::init ")
                             ACE_TEXT("couldn't open file for PERSISTENT ")
                             ACE_TEXT("data: %C\n"), file->name().c_str()));
                }
                continue;
              }

              DDS::Time_t timestamp;
              is >> timestamp.sec >> timestamp.nanosec >> std::noskipws;
              is.get(); // consume separator

              const size_t CHUNK = 4096;
              ACE_Message_Block mb(CHUNK);
              ACE_Message_Block * current = &mb;

              while (!is.eof()) {
                is.read(current->wr_ptr(), current->space());

                if (is.bad()) break;

                current->wr_ptr((size_t)is.gcount());

                if (current->space() == 0) {
                  ACE_Message_Block * old = current;
                  current = new ACE_Message_Block(CHUNK);
                  old->cont(current);
                }
              }

              sample_queue->enqueue_tail(
                sample_data_type(timestamp, mb, allocator));

              if (mb.cont()) mb.cont()->release();    // delete the cont() chain
            }
          }
        }
      }
    }
  }

  this->reactor_ = TheServiceParticipant->timer();
}

insert()

bool OpenDDS::DCPS::DataDurabilityCache::insert	(	DDS::DomainId_t	domain_id,
		char const *	topic_name,
		char const *	type_name,
		SendStateDataSampleList &	the_data,
		DDS::DurabilityServiceQosPolicy const &	qos
	)

Insert the samples corresponding to the given topic instance (uniquely identify by its domain, topic name and type name) into the data durability cache.

Definition at line 474 of file DataDurabilityCache.cpp.

References ACE_DEBUG, ACE_DES_FREE, ACE_ERROR, ACE_GUARD_RETURN, ACE_NEW_MALLOC_RETURN, ACE_SYNCH_MUTEX, ACE_TEXT(), allocator_, OpenDDS::DCPS::SendStateDataSampleList::begin(), ACE_Hash_Map_With_Allocator< class, class >::bind(), ACE_String_Base< char >::c_str(), cleanup_timer_ids_, OpenDDS::DCPS::COHERENT_CHANGE_FLAG, data_dir_, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::SendStateDataSampleList::end(), ACE_Hash_Map_With_Allocator< class, class >::find(), ACE_Allocator::free(), OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), OpenDDS::DCPS::DataSampleElement::get_sample(), OpenDDS::DCPS::DataDurabilityCache::sample_data_type::get_sample(), DDS::DurabilityServiceQosPolicy::history_depth, DDS::DurabilityServiceQosPolicy::history_kind, OpenDDS::DCPS::TimeDuration::is_zero(), DDS::KEEP_ALL_HISTORY_QOS, kind_, DDS::LENGTH_UNLIMITED, LM_DEBUG, LM_ERROR, lock_, ACE_Allocator::malloc(), DDS::DurabilityServiceQosPolicy::max_samples_per_instance, DDS::Time_t::nanosec, OPENDDS_STRING, OpenDDS::DCPS::OPENDDS_VECTOR(), DDS::PERSISTENT_DURABILITY_QOS, reactor_, samples_, ACE_Reactor_Timer_Interface::schedule_timer(), DDS::Time_t::sec, DDS::DurabilityServiceQosPolicy::service_cleanup_delay, OpenDDS::DCPS::SendStateDataSampleList::size(), OpenDDS::DCPS::DataSampleHeader::test_flag(), timestamp(), OpenDDS::DCPS::to_dds_string(), and OpenDDS::DCPS::TimeDuration::value().

Referenced by OpenDDS::DCPS::WriteDataContainer::persist_data().

{
  if (the_data.size() == 0)
    return true;  // Nothing to cache.

  // Apply DURABILITY_SERVICE QoS HISTORY and RESOURCE_LIMITS related
  // settings prior to data insertion into the cache.
  int depth = qos.history_kind == DDS::KEEP_ALL_HISTORY_QOS
    ? qos.max_samples_per_instance
    : qos.history_depth;

  if (depth == DDS::LENGTH_UNLIMITED)
    depth = 0x7fffffff;

  // Iterator to first DataSampleElement to be copied.
  SendStateDataSampleList::iterator element(the_data.begin());

  if (depth < 0)
    return false; // Should never occur.

  else if (depth == 0)
    return true;  // Nothing else to do.  Discard all data.

  else if (the_data.size() > depth) {
    // N.B. Dropping data samples does not take into account
    // those samples which are not actually persisted (i.e.
    // samples with the coherent_sample_ flag set). The spec
    // does not provide any guidance in this case, therefore
    // we opt for the simplest solution and assume that there
    // are no change sets when calculating the number of
    // samples to drop.

    // Drop "old" samples.  Only keep the "depth" most recent
    // samples, i.e. those found at the tail end of the
    // SendStateDataSampleList.
    ssize_t const advance_amount = the_data.size() - depth;
    std::advance(element, advance_amount);
  }

  // -----------

  // Copy samples to the domain/topic/type-specific cache.

  key_type const key(domain_id,
                     topic_name,
                     type_name,
                     this->allocator_.get());
  SendStateDataSampleList::iterator the_end(the_data.end());
  sample_list_type * sample_list = 0;

  typedef DurabilityQueue<sample_data_type> data_queue_type;
  data_queue_type ** slot = 0;
  data_queue_type * samples = 0;  // sample_list_type::value_type

  using OpenDDS::FileSystemStorage::Directory;
  using OpenDDS::FileSystemStorage::File;
  Directory::Ptr dir;
  OPENDDS_VECTOR(OPENDDS_STRING) path;
  {
    ACE_Allocator * const allocator = this->allocator_.get();

    ACE_GUARD_RETURN(ACE_SYNCH_MUTEX, guard, this->lock_, false);

    if (this->kind_ == DDS::PERSISTENT_DURABILITY_QOS) {
      try {
        dir = Directory::create(this->data_dir_.c_str());

        path.push_back(to_dds_string(domain_id));
        path.push_back(topic_name);
        path.push_back(type_name);
        dir = dir->get_dir(path);
        // dir is now the "type" directory, which is shared by all datawriters
        // of the domain/topic/type.  We actually need a new directory per
        // datawriter and this assumes that insert() is called once per
        // datawriter, as is currently the case.
        dir = dir->create_next_dir();
        path.push_back(dir->name());   // for use by the Cleanup_Handler

      } catch (const std::exception& ex) {
        if (DCPS_debug_level > 0) {
          ACE_ERROR((LM_ERROR,
                     ACE_TEXT("(%P|%t) DataDurabilityCache::insert ")
                     ACE_TEXT("couldn't create directory for PERSISTENT ")
                     ACE_TEXT("data: %C\n"), ex.what()));
        }

        dir.reset();
      }
    }

    if (this->samples_->find(key, sample_list, allocator) != 0) {
      // Create a new list (actually an ACE_Array_Base<>) with the
      // appropriate allocator passed to its constructor.
      ACE_NEW_MALLOC_RETURN(
        sample_list,
        static_cast<sample_list_type *>(
          allocator->malloc(sizeof(sample_list_type))),
        sample_list_type(1, static_cast<data_queue_type *>(0), allocator),
        false);

      if (this->samples_->bind(key, sample_list, allocator) != 0)
        return false;
    }

    data_queue_type ** const begin = &((*sample_list)[0]);
    data_queue_type ** const end =
      begin + sample_list->size();

    // Find an empty slot in the array.  This is a linear search but
    // that should be fine for the common case, i.e. a small number of
    // DataWriters that push data into the cache.
    slot = std::find(begin,
                     end,
                     static_cast<data_queue_type *>(0));

    if (slot == end) {
      // No available slots.  Grow the array accordingly.
      size_t const old_len = sample_list->size();
      sample_list->size(old_len + 1);

      data_queue_type ** new_begin = &((*sample_list)[0]);
      slot = new_begin + old_len;
    }

    ACE_NEW_MALLOC_RETURN(
      samples,
      static_cast<data_queue_type *>(
        allocator->malloc(sizeof(data_queue_type))),
      data_queue_type(allocator),
      false);

    // Insert the samples in to the sample list.
    *slot = samples;

    if (!dir.is_nil()) {
      samples->fs_path_ = path;
    }

    for (SendStateDataSampleList::iterator i(element); i != the_end; ++i) {
      DataSampleElement& elem = *i;

      // N.B. Do not persist samples with coherent changes.
      // To verify, we check the DataSampleHeader for the
      // coherent_change_ flag. The DataSampleHeader will
      // always be the first message block in the chain.
      //
      // It should be noted that persisting coherent changes
      // is a non-trivial task, and should be handled when
      // finalizing persistence profile conformance.
      if (DataSampleHeader::test_flag(COHERENT_CHANGE_FLAG, elem.get_sample())) {
        continue; // skip coherent sample
      }

      sample_data_type sample(elem, allocator);

      if (samples->enqueue_tail(sample) != 0)
        return false;

      if (!dir.is_nil()) {
        try {
          File::Ptr f = dir->create_next_file();
          std::ofstream os;

          if (!f->write(os)) return false;

          DDS::Time_t timestamp;
          const char * data;
          size_t len;
          sample.get_sample(data, len, timestamp);

          os << timestamp.sec << ' ' << timestamp.nanosec << ' ';
          os.write(data, len);

        } catch (const std::exception& ex) {
          if (DCPS_debug_level > 0) {
            ACE_ERROR((LM_ERROR,
                       ACE_TEXT("(%P|%t) DataDurabilityCache::insert ")
                       ACE_TEXT("couldn't write sample for PERSISTENT ")
                       ACE_TEXT("data: %C\n"), ex.what()));
          }
        }
      }
    }
  }

  // -----------

  // Schedule cleanup timer.
  //FUTURE: The cleanup delay needs to be persisted (if QoS is persistent)
  const TimeDuration cleanup_delay(qos.service_cleanup_delay);

  if (!cleanup_delay.is_zero()) {
    if (OpenDDS::DCPS::DCPS_debug_level >= 4) {
      ACE_DEBUG((LM_DEBUG,
                 ACE_TEXT("OpenDDS (%P|%t) Scheduling durable data ")
                 ACE_TEXT("cleanup for\n")
                 ACE_TEXT("OpenDDS (%P|%t) (domain_id, topic, type) ")
                 ACE_TEXT("== (%d, %C, %C)\n"),
                 domain_id,
                 topic_name,
                 type_name));
    }

    Cleanup_Handler * const cleanup =
      new Cleanup_Handler(*sample_list,
                          slot - &(*sample_list)[0],
                          this->allocator_.get(),
                          path,
                          this->data_dir_);
    ACE_Event_Handler_var safe_cleanup(cleanup);   // Transfer ownership
    long const tid =
      this->reactor_->schedule_timer(cleanup,
                                     0, // ACT
                                     cleanup_delay.value());
    if (tid == -1) {
      ACE_GUARD_RETURN(ACE_SYNCH_MUTEX, guard, this->lock_, false);

      ACE_DES_FREE(samples,
                   this->allocator_->free,
                   DurabilityQueue<sample_data_type>);
      *slot = 0;

      return false;

    } else {
      {
        ACE_GUARD_RETURN(ACE_SYNCH_MUTEX, guard, this->lock_, false);
        this->cleanup_timer_ids_.push_back(tid);
      }

      cleanup->timer_id(tid,
                        &this->cleanup_timer_ids_);
    }
  }

  return true;
}

OPENDDS_LIST()

typedef OpenDDS::DCPS::DataDurabilityCache::OPENDDS_LIST

(

long

)

operator=()

DataDurabilityCache& OpenDDS::DCPS::DataDurabilityCache::operator= ( DataDurabilityCache const &  )

private

Member Data Documentation

allocator_

unique_ptr<ACE_Allocator> const OpenDDS::DCPS::DataDurabilityCache::allocator_

private

Allocator used to allocate memory for sample map and lists.

Definition at line 227 of file DataDurabilityCache.h.

Referenced by get_data(), init(), insert(), and ~DataDurabilityCache().

cleanup_timer_ids_

timer_id_list_type OpenDDS::DCPS::DataDurabilityCache::cleanup_timer_ids_

private

Timer ID list.

Keep track of cleanup timer IDs in case we need to cancel before they expire.

Definition at line 241 of file DataDurabilityCache.h.

Referenced by insert(), and ~DataDurabilityCache().

data_dir_

ACE_CString OpenDDS::DCPS::DataDurabilityCache::data_dir_

private

Definition at line 231 of file DataDurabilityCache.h.

Referenced by init(), and insert().

kind_

DDS::DurabilityQosPolicyKind OpenDDS::DCPS::DataDurabilityCache::kind_

private

Definition at line 229 of file DataDurabilityCache.h.

Referenced by init(), and insert().

lock_

ACE_SYNCH_MUTEX OpenDDS::DCPS::DataDurabilityCache::lock_

private

Lock for synchronized access to the underlying map.

Definition at line 244 of file DataDurabilityCache.h.

Referenced by get_data(), and insert().

reactor_

ACE_Reactor_Timer_Interface* OpenDDS::DCPS::DataDurabilityCache::reactor_

private

Reactor with which cleanup timers will be registered.

Definition at line 247 of file DataDurabilityCache.h.

Referenced by init(), insert(), and ~DataDurabilityCache().

samples_

sample_map_type* OpenDDS::DCPS::DataDurabilityCache::samples_

private

Map of all data samples.

Definition at line 234 of file DataDurabilityCache.h.

Referenced by get_data(), init(), insert(), and ~DataDurabilityCache().

---

The documentation for this class was generated from the following files:

• DataDurabilityCache.h
• DataDurabilityCache.cpp