OpenDDS::DCPS::OwnershipManager Class Reference

#include <OwnershipManager.h>

List of all members.

Public Member Functions
typedef	OPENDDS_VECTOR (DataReaderImpl *) ReaderVec
typedef	OPENDDS_MAP (OPENDDS_STRING, InstanceMap) TypeInstanceMap
typedef	OPENDDS_VECTOR (WriterInfo) WriterInfos
typedef	OPENDDS_VECTOR (InstanceState *) 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, void *instance_map, DataReaderImpl *reader)
void *	get_instance_map (const char *type_name, DataReaderImpl *reader)
void	unregister_reader (const char *type_name, DataReaderImpl *reader)
void	remove_writer (const PublicationId &pub_id)
void	remove_writers (const ::DDS::InstanceHandle_t &instance_handle)
bool	remove_writer (const ::DDS::InstanceHandle_t &instance_handle, const PublicationId &pub_id)
bool	is_owner (const ::DDS::InstanceHandle_t &instance_handle, const PublicationId &pub_id)
bool	select_owner (const ::DDS::InstanceHandle_t &instance_handle, const PublicationId &pub_id, const CORBA::Long &ownership_strength, InstanceState *instance_state)
void	remove_owner (const ::DDS::InstanceHandle_t &instance_handle)
void	remove_instance (InstanceState *instance_state)
void	update_ownership_strength (const PublicationId &pub_id, const CORBA::Long &ownership_strength)
Private Member Functions
bool	remove_writer (const ::DDS::InstanceHandle_t &instance_handle, OwnershipWriterInfos &infos, const PublicationId &pub_id)
void	remove_owner (const ::DDS::InstanceHandle_t &instance_handle, OwnershipWriterInfos &infos, bool sort)
void	remove_candidate (OwnershipWriterInfos &infos, const PublicationId &pub_id)
void	broadcast_new_owner (const ::DDS::InstanceHandle_t &instance_handle, OwnershipWriterInfos &infos, const PublicationId &owner)
Private Attributes
ACE_Thread_Mutex	instance_lock_
TypeInstanceMap	type_instance_map_
InstanceOwnershipWriterInfos	instance_ownership_infos_
Classes
struct	InstanceMap
struct	OwnershipWriterInfos
struct	WriterInfo

---

Detailed Description

Definition at line 29 of file OwnershipManager.h.

---

Constructor & Destructor Documentation

OpenDDS::DCPS::OwnershipManager::OwnershipManager

(

)

Definition at line 36 of file OwnershipManager.cpp.

{
}

OpenDDS::DCPS::OwnershipManager::~OwnershipManager

(

)

Definition at line 41 of file OwnershipManager.cpp.

References type_instance_map_.

{
  // 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.
  TypeInstanceMap::iterator const the_end = type_instance_map_.end ();
  TypeInstanceMap::iterator iter = type_instance_map_.begin ();
  while (iter != the_end)
  {
    // There is no way to pass the instance map to concrete datareader
    // to delete, so it will be leaked.
    // delete iter->second.map_;
    ACE_DEBUG((LM_WARNING,
      ACE_TEXT("(%P|%t) OwnershipManager::~OwnershipManager ")
      ACE_TEXT("- non-empty type_instance_map_\n")));
    ++ iter;
  }

  type_instance_map_.clear ();
}

---

Member Function Documentation

void OpenDDS::DCPS::OwnershipManager::broadcast_new_owner	(	const ::DDS::InstanceHandle_t &	instance_handle,
		OwnershipWriterInfos &	infos,
		const PublicationId &	owner
	)			[private]

Definition at line 428 of file OwnershipManager.cpp.

References OpenDDS::DCPS::OwnershipManager::OwnershipWriterInfos::candidates_, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::OwnershipManager::OwnershipWriterInfos::instance_states_, OPENDDS_STRING, OpenDDS::DCPS::OwnershipManager::OwnershipWriterInfos::owner_, and OpenDDS::DCPS::OwnershipManager::WriterInfo::ownership_strength_.

Referenced by remove_owner(), and select_owner().

{
  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.
    GuidConverter writer_converter(owner);
    ACE_DEBUG((LM_DEBUG,
      ACE_TEXT("(%P|%t) OwnershipManager::broadcast_new_owner: ")
      ACE_TEXT("owner writer %C, instance handle %d strength %d num of candidates %d\n"),
      OPENDDS_STRING(writer_converter).c_str(),
      instance_handle, infos.owner_.ownership_strength_, infos.candidates_.size()));
  }

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

void * 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 76 of file OwnershipManager.cpp.

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

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

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

int OpenDDS::DCPS::OwnershipManager::instance_lock_acquire

(

)

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

Definition at line 64 of file OwnershipManager.cpp.

References instance_lock_.

{
  return this->instance_lock_.acquire ();
}

int OpenDDS::DCPS::OwnershipManager::instance_lock_release

(

)

Definition at line 70 of file OwnershipManager.cpp.

References instance_lock_.

{
  return this->instance_lock_.release ();
}

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

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

Definition at line 211 of file OwnershipManager.cpp.

References instance_ownership_infos_.

{
  ACE_GUARD_RETURN (ACE_Thread_Mutex,
                    guard,
                    this->instance_lock_,
                    false);

  InstanceOwnershipWriterInfos::iterator const the_end = instance_ownership_infos_.end ();

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

  return false;
}

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

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

typedef OpenDDS::DCPS::OwnershipManager::OPENDDS_VECTOR

(

InstanceState *

)

typedef OpenDDS::DCPS::OwnershipManager::OPENDDS_VECTOR

(

WriterInfo

)

typedef OpenDDS::DCPS::OwnershipManager::OPENDDS_VECTOR

(

DataReaderImpl *

)

void OpenDDS::DCPS::OwnershipManager::remove_candidate	(	OwnershipWriterInfos &	infos,
		const PublicationId &	pub_id
	)			[private]

Definition at line 297 of file OwnershipManager.cpp.

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

Referenced by remove_writer().

{
  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 candicate 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);
    }
  }
}

void OpenDDS::DCPS::OwnershipManager::remove_instance

(

InstanceState *

instance_state

)

Definition at line 154 of file OwnershipManager.cpp.

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

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

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

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

Definition at line 271 of file OwnershipManager.cpp.

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

{
  //change owner
  PublicationId new_owner(GUID_UNKNOWN);
  if (infos.candidates_.empty ()) {
    infos.owner_ = WriterInfo();
  }
  else {
    if (sort) {
      std::sort (infos.candidates_.begin(), infos.candidates_.end(),
                 ::Util::DescendingOwnershipStrengthSort);
    }

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

  this->broadcast_new_owner (instance_handle, infos, new_owner);
}

void OpenDDS::DCPS::OwnershipManager::remove_owner

(

const ::DDS::InstanceHandle_t &

instance_handle

)

Remove an owner of the specified instance.

Definition at line 452 of file OwnershipManager.cpp.

References instance_ownership_infos_.

Referenced by remove_writer(), and select_owner().

{
  ACE_GUARD(ACE_Thread_Mutex,
            guard,
            this->instance_lock_);

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

  this->remove_owner (instance_handle, iter->second, false);
}

bool OpenDDS::DCPS::OwnershipManager::remove_writer	(	const ::DDS::InstanceHandle_t &	instance_handle,
		OwnershipWriterInfos &	infos,
		const PublicationId &	pub_id
	)			[private]

Definition at line 253 of file OwnershipManager.cpp.

References OpenDDS::DCPS::OwnershipManager::OwnershipWriterInfos::owner_, OpenDDS::DCPS::OwnershipManager::WriterInfo::pub_id_, remove_candidate(), and remove_owner().

{
  if (infos.owner_.pub_id_ == pub_id) {
    this->remove_owner (instance_handle, infos, false);
    return true;
  }
  else {
    this->remove_candidate (infos, pub_id);
    return false;
  }

  return false;
}

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

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

Definition at line 232 of file OwnershipManager.cpp.

References instance_ownership_infos_, and remove_writer().

{
  ACE_GUARD_RETURN (ACE_Thread_Mutex,
                    guard,
                    this->instance_lock_,
                    false);

  InstanceOwnershipWriterInfos::iterator const the_end = instance_ownership_infos_.end ();

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

  return false;
}

void OpenDDS::DCPS::OwnershipManager::remove_writer

(

const PublicationId &

pub_id

)

Remove a writer from all instances ownership collection.

Definition at line 140 of file OwnershipManager.cpp.

References instance_ownership_infos_.

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

{
  ACE_GUARD(ACE_Thread_Mutex,
            guard,
            this->instance_lock_);

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

void OpenDDS::DCPS::OwnershipManager::remove_writers

(

const ::DDS::InstanceHandle_t &

instance_handle

)

Remove all writers that write to the specified instance.

Definition at line 170 of file OwnershipManager.cpp.

References OpenDDS::DCPS::DCPS_debug_level, and instance_ownership_infos_.

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

{
  InstanceStateVec instances_to_reset;
  {
    ACE_GUARD(ACE_Thread_Mutex,
              guard,
              this->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 ();
      InstanceStateVec::iterator const 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
  InstanceStateVec::iterator instance;
  for (instance  = instances_to_reset.begin();
       instance != instances_to_reset.end(); ++instance)
  {
    (*instance)->reset_ownership(instance_handle);
  }
}

bool OpenDDS::DCPS::OwnershipManager::select_owner	(	const ::DDS::InstanceHandle_t &	instance_handle,
		const PublicationId &	pub_id,
		const CORBA::Long &	ownership_strength,
		InstanceState *	instance_state
	)

Determine if the provided publication can be the owner.

Definition at line 321 of file OwnershipManager.cpp.

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

{
  ACE_GUARD_RETURN (ACE_Thread_Mutex,
                    guard,
                    this->instance_lock_,
                    false);

  InstanceOwnershipWriterInfos::iterator const the_end = instance_ownership_infos_.end ();

  InstanceOwnershipWriterInfos::iterator iter
    = instance_ownership_infos_.find (instance_handle);
  if (iter != the_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);
      this->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));
        this->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 currrent 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 candicate list. If not,
      // add it to the candidate list, otherwise update strength
      // if strength was changed.
      WriterInfos::iterator const 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.
        this->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);
    }
    this->broadcast_new_owner (instance_handle, infos, infos.owner_.pub_id_);
    return true;
  }

  return false;
}

void OpenDDS::DCPS::OwnershipManager::set_instance_map	(	const char *	type_name,
		void *	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 89 of file OwnershipManager.cpp.

References OpenDDS::DCPS::bind(), OpenDDS::DCPS::DCPS_debug_level, and type_instance_map_.

{
  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, 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 \"%s\"\n",type_name));
  }
}

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 106 of file OwnershipManager.cpp.

References OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::DataReaderImpl::delete_instance_map(), OpenDDS::DCPS::find(), OpenDDS::DCPS::OwnershipManager::InstanceMap::map_, OpenDDS::DCPS::OwnershipManager::InstanceMap::readers_, type_instance_map_, and OpenDDS::DCPS::unbind().

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

{
  ACE_GUARD(ACE_Thread_Mutex,
            guard,
            this->instance_lock_);

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

  ReaderVec::iterator end = instance->readers_.end();

  for (ReaderVec::iterator it(instance->readers_.begin());
      it != end; ++it) {
    if (*it == reader) {
      instance->readers_.erase (it);
      break;
    }
  }

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

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

Update the ownership strength of a publication.

---

Member Data Documentation

ACE_Thread_Mutex OpenDDS::DCPS::OwnershipManager::instance_lock_ [private]

Definition at line 165 of file OwnershipManager.h.

Referenced by instance_lock_acquire(), and instance_lock_release().

InstanceOwnershipWriterInfos OpenDDS::DCPS::OwnershipManager::instance_ownership_infos_ [private]

Definition at line 167 of file OwnershipManager.h.

Referenced by is_owner(), remove_instance(), remove_owner(), remove_writer(), remove_writers(), and select_owner().

TypeInstanceMap OpenDDS::DCPS::OwnershipManager::type_instance_map_ [private]

Definition at line 166 of file OwnershipManager.h.

Referenced by get_instance_map(), set_instance_map(), unregister_reader(), and ~OwnershipManager().

---

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

• OwnershipManager.h
• OwnershipManager.cpp

---