OpenDDS::DCPS::OwnershipManager Class Reference

#include <OwnershipManager.h>

Collaboration diagram for OpenDDS::DCPS::OwnershipManager:

Classes
struct	InstanceMap
struct	OwnershipWriterInfos
struct	WriterInfo

Public Member Functions
typedef	OPENDDS_SET (DataReaderImpl *) ReaderSet
typedef	OPENDDS_MAP (OPENDDS_STRING, InstanceMap) TypeInstanceMap
typedef	OPENDDS_VECTOR (WriterInfo) WriterInfos
typedef	OPENDDS_VECTOR (InstanceState_rch) InstanceStateVec
typedef	OPENDDS_MAP (DDS::InstanceHandle_t, OwnershipWriterInfos) InstanceOwnershipWriterInfos
	OwnershipManager ()
	~OwnershipManager ()
int	instance_lock_acquire ()
int	instance_lock_release ()
void	set_instance_map (const char *type_name, const RcHandle< RcObject > &instance_map, DataReaderImpl *reader)
RcHandle< RcObject >	get_instance_map (const char *type_name, DataReaderImpl *reader)
void	unregister_reader (const char *type_name, DataReaderImpl *reader)
void	remove_writer (const GUID_t &pub_id)
void	remove_writers (const DDS::InstanceHandle_t &instance_handle)
bool	remove_writer (const DDS::InstanceHandle_t &instance_handle, const GUID_t &pub_id)
bool	is_owner (const DDS::InstanceHandle_t &instance_handle, const GUID_t &pub_id)
bool	select_owner (const DDS::InstanceHandle_t &instance_handle, const GUID_t &pub_id, const CORBA::Long &ownership_strength, InstanceState_rch instance_state)
void	remove_owner (const DDS::InstanceHandle_t &instance_handle)
void	remove_instance (InstanceState *instance_state)
void	update_ownership_strength (const GUID_t &pub_id, const CORBA::Long &ownership_strength)

Private Member Functions
bool	remove_writer (const DDS::InstanceHandle_t &instance_handle, OwnershipWriterInfos &infos, const GUID_t &pub_id)
void	remove_owner (const DDS::InstanceHandle_t &instance_handle, OwnershipWriterInfos &infos, bool sort)
void	remove_candidate (OwnershipWriterInfos &infos, const GUID_t &pub_id)
void	broadcast_new_owner (const DDS::InstanceHandle_t &instance_handle, OwnershipWriterInfos &infos, const GUID_t &owner)

Private Attributes
ACE_Thread_Mutex	instance_lock_
TypeInstanceMap	type_instance_map_
InstanceOwnershipWriterInfos	instance_ownership_infos_

Detailed Description

Definition at line 31 of file OwnershipManager.h.

Constructor & Destructor Documentation

OwnershipManager()

OpenDDS::DCPS::OwnershipManager::OwnershipManager

(

)

Definition at line 33 of file OwnershipManager.cpp.

{
}

~OwnershipManager()

OpenDDS::DCPS::OwnershipManager::~OwnershipManager

(

)

Definition at line 37 of file OwnershipManager.cpp.

References ACE_DEBUG, ACE_TEXT(), and LM_WARNING.

{
  // The type->instance should be empty if unregister instance are performed
  // by all readers, but in case the instance not unregistered for some reason,
  // an error will be logged.
  if (!type_instance_map_.empty()) {
    // There is no way to pass the instance map to concrete datareader
    // to delete, so it will be leaked.
    ACE_DEBUG((LM_WARNING,
               ACE_TEXT("(%P|%t) OwnershipManager::~OwnershipManager ")
               ACE_TEXT("- non-empty type_instance_map_\n")));
  }
}

Member Function Documentation

broadcast_new_owner()

void OpenDDS::DCPS::OwnershipManager::broadcast_new_owner ( const DDS::InstanceHandle_t &  instance_handle, OwnershipWriterInfos &  infos, const GUID_t &  owner  )

private

Definition at line 385 of file OwnershipManager.cpp.

References ACE_DEBUG, ACE_TEXT(), OpenDDS::DCPS::OwnershipManager::OwnershipWriterInfos::candidates_, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::OwnershipManager::OwnershipWriterInfos::instance_states_, LM_DEBUG, OpenDDS::DCPS::OwnershipManager::OwnershipWriterInfos::owner_, and OpenDDS::DCPS::OwnershipManager::WriterInfo::ownership_strength_.

{
  if (DCPS_debug_level >= 1) {
    // This may not be an error since it could happen that the sample
    // is delivered to the datareader after the write is dis-associated
    // with this datareader.
    ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("(%P|%t) OwnershipManager::broadcast_new_owner: ")
               ACE_TEXT("owner writer %C, instance handle %d strength %d num ")
               ACE_TEXT("of candidates %d\n"),
               LogGuid(owner).c_str(), instance_handle,
               infos.owner_.ownership_strength_,
               (int)infos.candidates_.size()));
  }

  const InstanceStateVec::iterator the_end = infos.instance_states_.end();
  for (InstanceStateVec::iterator iter = infos.instance_states_.begin();
       iter != the_end; ++iter) {
    (*iter)->set_owner(owner);
  }
}

get_instance_map()

RcHandle< RcObject > OpenDDS::DCPS::OwnershipManager::get_instance_map	(	const char *	type_name,
		DataReaderImpl *	reader
	)

Accesor of the instance map for provided type. It is called once for each new instance in a datareader.

Definition at line 64 of file OwnershipManager.cpp.

References OpenDDS::DCPS::find(), OpenDDS::DCPS::OwnershipManager::InstanceMap::map_, and OpenDDS::DCPS::OwnershipManager::InstanceMap::readers_.

Referenced by OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::store_instance_data().

{
  InstanceMap* instance = 0;
  if (0 != find(type_instance_map_, type_name, instance)) {
    return RcHandle<RcObject>();
  }

  instance->readers_.insert(reader);
  return instance->map_;
}

instance_lock_acquire()

int OpenDDS::DCPS::OwnershipManager::instance_lock_acquire

(

)

Acquire/release lock for type instance map. The following functions are synchronized by instance_lock_.

Definition at line 52 of file OwnershipManager.cpp.

{
  return instance_lock_.acquire();
}

instance_lock_release()

int OpenDDS::DCPS::OwnershipManager::instance_lock_release

(

)

Definition at line 58 of file OwnershipManager.cpp.

{
  return instance_lock_.release();
}

is_owner()

bool OpenDDS::DCPS::OwnershipManager::is_owner	(	const DDS::InstanceHandle_t &	instance_handle,
		const GUID_t &	pub_id
	)

Return true if the provide writer is the owner of the instance.

Definition at line 187 of file OwnershipManager.cpp.

References ACE_GUARD_RETURN.

Referenced by OpenDDS::DCPS::DataReaderImpl::data_received(), and OpenDDS::DCPS::InstanceState::dispose_was_received().

{
  ACE_GUARD_RETURN(ACE_Thread_Mutex, guard, instance_lock_, false);

  InstanceOwnershipWriterInfos::iterator iter
    = instance_ownership_infos_.find(instance_handle);
  if (iter != instance_ownership_infos_.end()) {
    return iter->second.owner_.pub_id_ == pub_id;
  }

  return false;
}

OPENDDS_MAP() [1/2]

typedef OpenDDS::DCPS::OwnershipManager::OPENDDS_MAP	(	OPENDDS_STRING	,
		InstanceMap
	)

OPENDDS_MAP() [2/2]

typedef OpenDDS::DCPS::OwnershipManager::OPENDDS_MAP	(	DDS::InstanceHandle_t	,
		OwnershipWriterInfos
	)

OPENDDS_SET()

typedef OpenDDS::DCPS::OwnershipManager::OPENDDS_SET

(

DataReaderImpl *

)

OPENDDS_VECTOR() [1/2]

typedef OpenDDS::DCPS::OwnershipManager::OPENDDS_VECTOR

(

WriterInfo

)

OPENDDS_VECTOR() [2/2]

typedef OpenDDS::DCPS::OwnershipManager::OPENDDS_VECTOR

(

InstanceState_rch

)

remove_candidate()

void OpenDDS::DCPS::OwnershipManager::remove_candidate ( OwnershipWriterInfos &  infos, const GUID_t &  pub_id  )

private

Definition at line 260 of file OwnershipManager.cpp.

References OpenDDS::DCPS::OwnershipManager::OwnershipWriterInfos::candidates_.

{
  if (!infos.candidates_.empty()) {
    WriterInfos::iterator const the_end = infos.candidates_.end();

    WriterInfos::iterator found_candidate = the_end;
    // Supplied writer is not an owner, check if it exists in candidate list.
    // If not, add it to the candidate list and sort the list.
    for (WriterInfos::iterator iter = infos.candidates_.begin();
         iter != the_end; ++iter) {
      if (iter->pub_id_ == pub_id) {
        found_candidate = iter;
        break;
      }
    }

    if (found_candidate != the_end) {
      infos.candidates_.erase(found_candidate);
    }
  }
}

remove_instance()

void OpenDDS::DCPS::OwnershipManager::remove_instance

(

InstanceState *

instance_state

)

Definition at line 132 of file OwnershipManager.cpp.

References ACE_GUARD, and OpenDDS::DCPS::InstanceState::instance_handle().

Referenced by OpenDDS::DCPS::InstanceState::~InstanceState().

{
  ACE_GUARD(ACE_Thread_Mutex, guard, instance_lock_);
  const DDS::InstanceHandle_t ih = instance_state->instance_handle();
  InstanceOwnershipWriterInfos::iterator i = instance_ownership_infos_.find(ih);
  if (i != instance_ownership_infos_.end()) {
    InstanceStateVec& states = i->second.instance_states_;
    for (size_t j = 0; j < states.size(); ++j) {
      if (states[j].in() == instance_state) {
        states.erase(states.begin() + j);
        break;
      }
    }
  }
}

remove_owner() [1/2]

void OpenDDS::DCPS::OwnershipManager::remove_owner

(

const DDS::InstanceHandle_t &

instance_handle

)

Remove an owner of the specified instance.

Definition at line 410 of file OwnershipManager.cpp.

References ACE_GUARD, and OPENDDS_END_VERSIONED_NAMESPACE_DECL.

{
  ACE_GUARD(ACE_Thread_Mutex, guard, instance_lock_);

  const InstanceOwnershipWriterInfos::iterator iter =
    instance_ownership_infos_.find(instance_handle);

  if (iter != instance_ownership_infos_.end()) {
    remove_owner(instance_handle, iter->second, false);
  }
}

remove_owner() [2/2]

void OpenDDS::DCPS::OwnershipManager::remove_owner ( const DDS::InstanceHandle_t &  instance_handle, OwnershipWriterInfos &  infos, bool  sort  )

private

Definition at line 234 of file OwnershipManager.cpp.

References OpenDDS::DCPS::OwnershipManager::OwnershipWriterInfos::candidates_, OpenDDS::DCPS::Util::DescendingOwnershipStrengthSort(), OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::OwnershipManager::OwnershipWriterInfos::owner_, and OpenDDS::DCPS::OwnershipManager::WriterInfo::pub_id_.

{
  //change owner
  GUID_t new_owner(GUID_UNKNOWN);
  if (infos.candidates_.empty()) {
    infos.owner_ = WriterInfo();

  } else {
    if (sort) {
      std::sort(infos.candidates_.begin(), infos.candidates_.end(),
                Util::DescendingOwnershipStrengthSort);
    }

    const WriterInfos::iterator begin = infos.candidates_.begin();
    infos.owner_ = *begin;
    infos.candidates_.erase(begin);
    new_owner = infos.owner_.pub_id_;
  }

  broadcast_new_owner(instance_handle, infos, new_owner);
}

remove_writer() [1/3]

void OpenDDS::DCPS::OwnershipManager::remove_writer

(

const GUID_t &

pub_id

)

Remove a writer from all instances ownership collection.

Definition at line 119 of file OwnershipManager.cpp.

References ACE_GUARD.

Referenced by OpenDDS::DCPS::InstanceState::unregister_was_received(), OpenDDS::DCPS::DataReaderImpl::writer_became_dead(), and OpenDDS::DCPS::DataReaderImpl::writer_removed().

{
  ACE_GUARD(ACE_Thread_Mutex, guard, instance_lock_);

  const InstanceOwnershipWriterInfos::iterator the_end =
    instance_ownership_infos_.end();
  for (InstanceOwnershipWriterInfos::iterator iter =
         instance_ownership_infos_.begin(); iter != the_end; ++iter) {
    remove_writer(iter->first, iter->second, pub_id);
  }
}

remove_writer() [2/3]

bool OpenDDS::DCPS::OwnershipManager::remove_writer	(	const DDS::InstanceHandle_t &	instance_handle,
		const GUID_t &	pub_id
	)

Remove a writer that write to the specified instance. Return true if it's the owner writer removed.

Definition at line 203 of file OwnershipManager.cpp.

References ACE_GUARD_RETURN.

{
  ACE_GUARD_RETURN(ACE_Thread_Mutex, guard, instance_lock_, false);

  InstanceOwnershipWriterInfos::iterator the_iter =
    instance_ownership_infos_.find(instance_handle);
  if (the_iter != instance_ownership_infos_.end()) {
    return remove_writer(instance_handle, the_iter->second, pub_id);
  }

  return false;
}

remove_writer() [3/3]

bool OpenDDS::DCPS::OwnershipManager::remove_writer ( const DDS::InstanceHandle_t &  instance_handle, OwnershipWriterInfos &  infos, const GUID_t &  pub_id  )

private

Definition at line 218 of file OwnershipManager.cpp.

References OpenDDS::DCPS::OwnershipManager::OwnershipWriterInfos::owner_, and OpenDDS::DCPS::OwnershipManager::WriterInfo::pub_id_.

{
  if (infos.owner_.pub_id_ == pub_id) {
    remove_owner(instance_handle, infos, false);
    return true;

  } else {
    remove_candidate(infos, pub_id);
    return false;
  }
}

remove_writers()

void OpenDDS::DCPS::OwnershipManager::remove_writers

(

const DDS::InstanceHandle_t &

instance_handle

)

Remove all writers that write to the specified instance.

Definition at line 149 of file OwnershipManager.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, and LM_DEBUG.

Referenced by OpenDDS::DCPS::DataReaderImpl::process_deadline(), and OpenDDS::DCPS::DataReaderImpl::release_instance().

{
  InstanceStateVec instances_to_reset;
  {
    ACE_GUARD(ACE_Thread_Mutex, guard, instance_lock_);

    if (DCPS_debug_level >= 1) {
      ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) OwnershipManager::remove_writers:")
                 ACE_TEXT(" disassociate writers with instance %d\n"),
                 instance_handle));
    }

    InstanceOwnershipWriterInfos::iterator owner_wi =
      instance_ownership_infos_.find(instance_handle);
    if (owner_wi != instance_ownership_infos_.end()) {
      owner_wi->second.owner_ = WriterInfo();
      owner_wi->second.candidates_.clear();
      const InstanceStateVec::iterator end =
        owner_wi->second.instance_states_.end();
      for (InstanceStateVec::iterator iter =
             owner_wi->second.instance_states_.begin(); iter != end; ++iter) {
        // call after lock released, will call back to data reader
        instances_to_reset.push_back(*iter);
      }
      owner_wi->second.instance_states_.clear();

      instance_ownership_infos_.erase(owner_wi);
    }
  }
  // Lock released
  for (InstanceStateVec::iterator instance = instances_to_reset.begin();
       instance != instances_to_reset.end(); ++instance) {
    (*instance)->reset_ownership(instance_handle);
  }
}

select_owner()

bool OpenDDS::DCPS::OwnershipManager::select_owner	(	const DDS::InstanceHandle_t &	instance_handle,
		const GUID_t &	pub_id,
		const CORBA::Long &	ownership_strength,
		InstanceState_rch	instance_state
	)

Determine if the provided publication can be the owner.

Definition at line 284 of file OwnershipManager.cpp.

References ACE_GUARD_RETURN, OpenDDS::DCPS::OwnershipManager::OwnershipWriterInfos::candidates_, OpenDDS::DCPS::Util::DescendingOwnershipStrengthSort(), OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::OwnershipManager::OwnershipWriterInfos::instance_states_, OpenDDS::DCPS::OwnershipManager::OwnershipWriterInfos::owner_, OpenDDS::DCPS::OwnershipManager::WriterInfo::ownership_strength_, OpenDDS::DCPS::OwnershipManager::WriterInfo::pub_id_, and OpenDDS::DCPS::InstanceState::registered().

Referenced by OpenDDS::DCPS::DataReaderImpl::ownership_filter_instance().

{
  ACE_GUARD_RETURN(ACE_Thread_Mutex, guard, instance_lock_, false);

  InstanceOwnershipWriterInfos::iterator iter =
    instance_ownership_infos_.find(instance_handle);
  if (iter != instance_ownership_infos_.end()) {
    OwnershipWriterInfos& infos = iter->second;
    if (!instance_state->registered()) {
      infos.instance_states_.push_back(instance_state);
      instance_state->registered(true);
    }

    // No owner at some point.
    if (infos.owner_.pub_id_ == GUID_UNKNOWN) {
      infos.owner_ = WriterInfo(pub_id, ownership_strength);
      broadcast_new_owner(instance_handle, infos, pub_id);
      return true;

    } else if (infos.owner_.pub_id_ == pub_id) { // is current owner
      //still owner but strength changed to be bigger..
      if (infos.owner_.ownership_strength_ <= ownership_strength) {
        infos.owner_.ownership_strength_ = ownership_strength;
        return true;

      } else { //update strength and reevaluate owner which broadcast new owner.
        infos.candidates_.push_back(WriterInfo(pub_id, ownership_strength));
        remove_owner(instance_handle, infos, true);
        return infos.owner_.pub_id_ == pub_id;
      }

    } else { // not current owner, reevaluate the owner
      bool replace_owner = false;
      // Add current owner to candidate list for owner reevaluation
      // if provided pub has strength greater than current owner.
      if (ownership_strength > infos.owner_.ownership_strength_) {
        infos.candidates_.push_back(infos.owner_);
        replace_owner = true;
      }

      bool found = false;
      bool sort = true;

      // check if it already existed in candidate list. If not,
      // add it to the candidate list, otherwise update strength
      // if strength was changed.
      const WriterInfos::iterator the_end = infos.candidates_.end();

      for (WriterInfos::iterator iter = infos.candidates_.begin();
           iter != the_end; ++iter) {

        if (iter->pub_id_ == pub_id) {
          if (iter->ownership_strength_ != ownership_strength) {
            iter->ownership_strength_ = ownership_strength;
          } else {
            sort = false;
          }
          found = true;
          break;
        }
      }

      if (!found) {
        infos.candidates_.push_back(WriterInfo(pub_id, ownership_strength));
      }

      if (sort) {
        std::sort(infos.candidates_.begin(), infos.candidates_.end(),
                  Util::DescendingOwnershipStrengthSort);
      }

      if (replace_owner) {
        // Owner was already moved to candidate list and the list was sorted
        // already so pick owner from sorted list and replace current
        // owner.
        remove_owner(instance_handle, infos, false);
      }

      return infos.owner_.pub_id_ == pub_id;
    }

  } else {
    // first writer of the instance so it's owner.
    OwnershipWriterInfos& infos = instance_ownership_infos_[instance_handle];
    infos.owner_ = WriterInfo(pub_id, ownership_strength);
    if (!instance_state->registered()) {
      infos.instance_states_.push_back(instance_state);
      instance_state->registered(true);
    }
    broadcast_new_owner(instance_handle, infos, infos.owner_.pub_id_);
    return true;
  }

  return false;
}

set_instance_map()

void OpenDDS::DCPS::OwnershipManager::set_instance_map	(	const char *	type_name,
		const RcHandle< RcObject > &	instance_map,
		DataReaderImpl *	reader
	)

The instance map per type is created by the concrete datareader when first sample with the type is received.

Definition at line 77 of file OwnershipManager.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::bind(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::RcHandle< T >::in(), LM_DEBUG, and LM_ERROR.

Referenced by OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::store_instance_data().

{
  if (DCPS_debug_level >= 1) {
    ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) OwnershipManager::set_instance_map ")
               ACE_TEXT("instance map %X is created by reader %X\n"),
               instance_map.in(), reader));
  }

  if (0 != OpenDDS::DCPS::bind(type_instance_map_, type_name,
                               InstanceMap(instance_map, reader))) {
    ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: OwnershipManager::set_instance_map "
               "failed to bind instance for type \"%C\"\n", type_name));
  }
}

unregister_reader()

void OpenDDS::DCPS::OwnershipManager::unregister_reader	(	const char *	type_name,
		DataReaderImpl *	reader
	)

The readers that access the instance map are keep tracked as ref counting to the instance map. The readers need unregister itself with the instance map upon unregistering instance.The instance map is deleted upon the last reader unregistering an instance of the type.

Definition at line 95 of file OwnershipManager.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::find(), LM_DEBUG, OpenDDS::DCPS::OwnershipManager::InstanceMap::map_, OpenDDS::DCPS::OwnershipManager::InstanceMap::readers_, and OpenDDS::DCPS::unbind().

Referenced by OpenDDS::DCPS::DataReaderImpl::cleanup().

{
  ACE_GUARD(ACE_Thread_Mutex, guard, instance_lock_);

  InstanceMap* instance = 0;
  if (0 != find(type_instance_map_, type_name, instance)) {
    return;
  }

  instance->readers_.erase(reader);

  if (instance->readers_.empty()) {
    if (DCPS_debug_level >= 1) {
      ACE_DEBUG((LM_DEBUG,
                 ACE_TEXT("(%P|%t) OwnershipManager::unregister_reader ")
                 ACE_TEXT(" instance map %@ is deleted by reader %@\n"),
                 instance->map_.in(), reader));
    }
    unbind(type_instance_map_, type_name);
  }
}

update_ownership_strength()

void OpenDDS::DCPS::OwnershipManager::update_ownership_strength	(	const GUID_t &	pub_id,
		const CORBA::Long &	ownership_strength
	)

Update the ownership strength of a publication.

Member Data Documentation

instance_lock_

ACE_Thread_Mutex OpenDDS::DCPS::OwnershipManager::instance_lock_

private

Definition at line 173 of file OwnershipManager.h.

instance_ownership_infos_

InstanceOwnershipWriterInfos OpenDDS::DCPS::OwnershipManager::instance_ownership_infos_

private

Definition at line 175 of file OwnershipManager.h.

type_instance_map_

TypeInstanceMap OpenDDS::DCPS::OwnershipManager::type_instance_map_

private

Definition at line 174 of file OwnershipManager.h.

---

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

• OwnershipManager.h
• OwnershipManager.cpp