DDS Lifespan QoS - Yaşam Süresi Yani Expiration Time

Giriş

Lifespan QoS verinin yaşam süresini belirtiyor. Veri güncellendiği müddetçe yaşam süresi yenileniyor, ancak güncellenmezse ve belirtilen yaşam süresi biterse DDS önbelleğinden siliniyor. Açıklaması şöyle

The purpose of this QoS is to avoid delivering "stale" data to the application.

Each data sample written by the DataWriter has an associated 'expiration time' beyond which the data should not be delivered to any application. Once the sample expires, the data will be removed from the DataReader caches as well as from the transient and persistent information caches.

...

This QoS relies on the sender and receiving applications having their clocks sufficiently synchronized. If this is not the case and the Service can detect it, the </blockquote>is allowed to use the reception timestamp instead of the source timestamp in its computation of the 'expiration time'

DataWriter Açısından

Gönderilmek üzere DataWriter belleğinde bekleyen yaşam süresi dolmuş veriler, DDS tarafından silinirler. Böylece eski veri dağıtılmayarak kaynak kullanımına katkıda bulunulur.

DataReader Açısından

DataReader tarafından henüz okunmamış ancak yaşam süresi dolmuş veriler, DDS tarafından silinirler. 

Örnek

Örneğin taktik veri linklerinde bir iz belli bir süre boyunca güncellenmezse o iz sistemden düşürülür. DDS ile iz kuyruğuna Lifespan QoS parametresi atanırsa, izin temizlenme işi DDS tarafından otomatik olarak yapılır.

DDS TimeBasedFilter Qos - DataReader İçin Geçerlidir

Giriş

İki tane read() işlemi arasında olması istenen "minimum separation" süresi tanımlanabilir. Böylece eğer DataWriter çok fazla veri gönderiyorsa, bir kısmı süzülür

DDS Ownership Qos - DataWriter İçin Geçerlidir

Giriş

Aynı topic bilgisini gönderen birden fazla publisher varsa, hangisini tercih edileceğini belirtir. OwnershipQosPolicy yapısı kullanılır

kind Alanı

Bir enum'dur. Açıklaması şöyle

OwnershipQosPolicyKind

There are two possible values (see OwnershipQosPolicyKind):

SHARED_OWNERSHIP_QOS: This option indicates that the service does not enforce unique ownership for each instance. In this case, multiple DataWriters are allowed to update the same data instance and all the updates are made available to the existing DataReaders. Those updates are also subject to the TimeBasedFilterQosPolicy or HistoryQosPolicy settings, so they can be filtered.

EXCLUSIVE_OWNERSHIP_QOS: This option indicates that each instance can only be updated by one DataWriter, meaning that at any point in time a single DataWriter owns each instance and is the only one whose modifications will be visible for the existing DataReaders. The owner can be changed dynamically according to the highest strength between the alive DataWriters, which has not violated the deadline contract concerning the data instances. That strength can be changed using the OwnershipStrengthQosPolicy.

DDS DomainParticipantQos Sınıfı

participant_discovery_protocol Alanı

Discovery için kullanılacak multicast adresi belirtir.

Örnek

Şöyle yaparız

DomainParticipantQos qos = ...;
qos.participant_discovery_protocol.add("239.255.0.50");

Daha sonra şöyle yaparız

DomainParticipantFactory dpf = ...;
dpf.set_default_pariticipant_qos(qos);

Data Distribution Service (DDS) Nerede Kullanılır ve Kullanılmaz

Giriş

Bu yazıda bazı genel prensipler var. Bu prensipler bana "doğal" gelen kullanımlara göre yazıldı. Bazı durumlarda farklı teknolojileri kullanarak aynı sonuca erişmek mümkün olsa da, "doğal" hissiyatını vermiyor. 

DDS Hangi İşlere Uygun Değil

Bence DDS şu durumlara uygun değil

1. Merkezi bir broker gerektiren durum

2. Event Replay gerektiren durum

3. Consumer Group gerektiren durum

4. Çok yoğun mesaj geldiği durum

5. Remote Procedure Call gerektiren durum

1. Merkezi Broker Gerektiren Durum - Mesaj Kaybı Olmamalı

Genellikle bir event'in mutlaka bunu işleyen tüketiciye ulaşması durumudur. Tüketici taraf event'in oluştuğu anda, çalışıyor veya çalışmıyor olabilir. 

Örneğin bir para transferi yapmak isteyelim. Burada para örneğini kaybolursa çok yaygara çıkacağı için seçtim :) Merkezi bir broker varsa - mesela RabbitMQ - bu broker üzerindeki durable (kalıcı) kuyruklar sayesinde, hem kuyruğun kendisini, hem de içindeki event'leri saklayabiliyor.

Tüketici tarafın mesajı işlediğine dair acknowledgement (onay) gelinceye kadar, mesaj broker'da saklanabilir, hata oluşması durumunda tekrar gönderilebilir. 

DDS mesaj kaybı olmayacağını "Durability Service" kullanılmıyorsa tam olarak garanti edemiyor.

"Durability Service" kullanılmayan bir durumda, eğer DataWriter ve DataReader birbirlerini bulmadıysa (discover) veya eşleşmediyse mesajlar rahatlıkla kaybolabilir. Ve hatta discovery için bir miktar da beklemek gerekebilir. DataWriter yazına bakabilirsiniz.

2. Transaction Gerektiren Durum

Açıklaması şöyle

JMS offers some capabilities not offered by DDS. Distinctive JMS capabilities include point-to-point delivery to exactly one of many consumers, message priority, and enterprise specific features such as full transactional support, and application level acknowledgements. 

3. Event Replay Gerektiren Durum

Event replay gerektiren durumlar, genellikle bir merkezi broker'ın event'leri belli bir süre boyunca saklaması şeklinde çözülüyor. Örneğin Apache Kafka bu şekilde çalışıyor. Eğer saklama süresi sonsuz olarak seçilirse, aslında sürekli ekleme yapılan bir veri tabanına dönüşebilir. Bu teknoloji Time Series Database olarak anılıyor

4. Consumer Group Gerektiren Durum

Consumer group, elimizde çok fazla sayıda event varsa ve bu event'leri tek bir uygulama ile tüketemiyorsak gerekiyor. Amacımız bu event'leri tüm tüketicilere dağıtmak. Yine Apache Kafka ve benzeri teknolojiler Consumer Group oluşturmaya, bu grubun otomatik olarak yönetilmesini imkan tanıyor. 

5. Çok yoğun mesaj Gerektiren Durum

Dakikada gelen mesaj sayısı 20K-30K ve üstü ise bana göre yoğun mesaj geliyordur. Bu sayılar benim gözlemlerim. Yoğun mesajları işleyebilmek için, sharding yapmak gerekiyor. Yani yükün birden fazla broker'a dağıtılması. 

5. Remote Procedure Call Gerektiren Durum

 Remote Message Call (RPC) karşı sisteme bir çağrı yapmak ve cevap beklemek anlamına gelir. Dağıtık ve senkron çalışan sitemlerde RPC yapmak gerekiyor. Asenkron çalışan mesaj kuyrukları bu işlevi yerine getirebiliyor. Örneğin JMS broker'ları bu yöntemi destekliyor.

DDS Hangi İşlere Uygun

Bence DDS şu durumlara uygun

1. Mesaj broadcast edildiği durum

2. QoS parametresi gerektiği durum

1. Mesaj Broadcast Edildiği Durum

Bu kullanım şekline özellikle "broadcast" dedim ama "fire and forget" olarak ta isimlendirilebilir. 

Elimizde bir alıcı yani sensör olsun. Bu alıcı ürettiği bilgiyi belli aralıklarda yayınlıyor olsun. Kapalı bir sistemde - gemi, uçak vs - bir çok başka birim bu alıcının gönderdiği bilgiye ihtiyaç duyar. 

Bu gibi sistemlerde merkezi bir broker kullanılmak istenmeyebilir. Çünkü broker sistemin tam kalbine yerleşecek ve "Single Point of Failure" oluşturacaktır. 

Bu kullanım şeklinde iletişim yine merkezi bir broker olmaması yüzünden, daha da hızlı olacaktır. Zaten bu yüzden DDS için "Near Realtime" deniliyor. Tabii Realtime çok genel bir gelime :) 

Ancak buradaki vurgu iletişimde, broker'ın getirdiği maliyetlerin azalması.

2. QoS Parametresi Gerekiyorsa

DDS'in tanımladığı bir sürü QoS var. Eğer bunlardan bir tanesi lazımsa, tabii ki DDS kolay çözümler sunabiliyor. 

En çok kullanılan QoS parametrelerinden bir tanesi bence DDS'in "Sampling" yapabilme yeteneği. Bu yetenek "History QoS" olarak biliniyor.  Örneğin her "key" değeri için kaç tane "sample" saklamak istediğimizi belirtmek ve bu key'in geçmişine gidebilmek kolaylaşıyor.

Diğer güzel QoS'ler Deadline Qos ve Liveliness Qos.  Yukarıdaki alıcı örneğinde alıcıların yedeklenmesi primary sensor, secondary sensor şeklinde çalışabilme imkanı elde edilebilir. 

DDS QoS Parametreleri Listesi - Quality of Service

Giriş
DDS bir çok QoS (Quality of Service) parametresini destekler. Bunlar şöyle

History

Depth

Reliability

Durability

Deadline

Liveliness

Lease Duration

Durability QoS

Durability Qos yazısına taşıdım

History QoS
History Qos yazısına taşıdım.

Lifespan QoS

Lifespan QoS yazısına taşıdım.

Ownership QoS
Ownership Qos yazısına taşıdım.

Resource Limits QoS

Resource Limits Qos yazısına taşıdım

Deadline QoS ve Liveliness QoS
Deadline Qos ve Liveliness Qos yazısına taşıdım

TimeBasedFilter Qos

TimeBasedFilter Qos yazısına taşıdım

TopicQoS

Şu QoS parametrelerine sahiptir

TopicData

Durability

DurabilityService

Deadline

LatencyBudget

Liveliness

Reliability

DestinationOrder

History

ResourceLimits

TransportPriority

Lifespan

Ownership

DataRepresentation

ReaderTransport

DataReaderQoS

Şu QoS parametrelerine sahiptir

Durability

Deadline

LatencyBudget

Liveliness

Reliability

DestinationOrder

History

ResourceLimits

UserData

Ownership

TimeBasedFilter

ReaderDataLifecycle

DataRepresentation

TypeConsistencyEnforcement

Property

ReaderTransportProtocolPolicy

EntityId

DataWriterQoS

Şu QoS parametrelerine sahiptir

Durability

DurabilityService

Deadline

LatencyBudget

Liveliness

Reliability

DestinationOrder

History

ResourceLimits

TransportPriority

Lifespan

UserData

Ownership

OwnershipStrength

WriterDataLifecycle

DataRepresentation

Property

WriterTransportProtocolPolicy

EntityId

Publisher QoS

Şu QoS parametrelerine sahiptir

Presentation

Partition

GroupData

EntityFactory

Subsriber QoS

Şu QoS parametrelerine sahiptir

Presentation

Partition

GroupData

EntityFactory

DDS Durability Qos - Kalıcılık

Durability Service Nedir

Durability kelimesi Kalıcılık anlamına gelir. "Durability Service" Persistent veya Transient olarak ayarlanmış ise geç gelen (late joining) bir DataReader'ın kaç tane geçmiş veri ile beslenip güncel duruma getirilmesi gerektiğini belirtir.

OpenSplice
OpenSplice için  "Durability Service" ayarları ve sonuçları şöyle

- PersistentDurabilityQos           : Late joiner gets history data
- TransientDurabilityQos           :  Late joiner gets history data
- TransientLocalDurabilityQos   : Late joiner gets history data
- VolatileDurabilityQos               : Late joiner does not get history data.

1. PersistentDurability Nedir? - Diskte Kalıcılık
Delegated Durability kullanır. Veri harici bir DurabilityService üzerinde ve diskte saklanır. 

Yani DataWriter'ın konuştuğu kelimeler DurabilityService tarafından diske yazılır. DurabilityService ölüp tekrar çalışsa bile, veriyi diskten okur ve tekrar hafızasına alır. Uygulamanın konfigürasyon parametrelerini saklamak için kullanılabilir.

2. TransientDurability Nedir?

Delegated Durability kullanır. Veri harici bir DurabilityService üzerinde saklanır. DurabilityService tüm veriyi hafızasında tutmaktadır.

Geç gelen bir DataReader, odaya girince DurabilityService'ten elindeki verileri göndermesini ister. Böylece daha önce konuşulmuş kelimeleri öğrenir. Kelimeleri konuşmuş olan DataWriter hayatta olmasa bile konuşmaları kaydeden DurabilityService hayatta olduğu için geçmişi öğrenmek mümkündür. 

3. TransientLocalDurability Nedir?

Veri bu veriyi üretmiş olan DataWriter üzerinde saklanır. 

Geç gelen bir DataReader, odaya girince bağlandığı DataWriter ona belleğindeki verileri aktarır. Yani daha önce konuşulmuş kelimeleri duymak için DataWriter'ın hayatta olmuş olması gerekir.

4. VolatileDurability Nedir ?
Volatile kelimesi Uçucu anlamına gelir. Veri saklanmaz!

Geç kelen bir DataReader, odada daha önce konuşulmuş olan, kelimeleri duyamaz. Çünkü ona söyleyecek kimse yoktur. VolatileDurability de benzer şekilde çalışır. Geç gelen bir reader daha önceki veriyi alamaz. 

Veriye Göre Durability Seçimi

Bu konuyla ilgili bir yazı burada.

1. Sürekli Akan Veri

Mesela sürekli veri akan durumda VolatileDurability kullanılabilir, çünkü eski verinin anlamı yoktur, yerine yenisi gelecektir. Bu durumda bazı Qos parametreleri şöyle olur

Reliability : Best Effort. Sürekli akan veri UDP gibi çalışabilir

Durability : Volatile

History : Keep last

Deadline : İstenilen bir değer. Sürekli akan veri belli bir süre kesilirse tetiklenir

Time Based Filter : Minimum time separation belirtilerek selective update yapılabilir.

2. State Verisi

Uygulama tarafından kullanılan durum verisi. Bu veri az güncellendiği için geç gelen katılımcıya son state verisini vermek gerekir. State verisini diskte saklamaya gerek yoktur.

Reliability : Reliable. 

Durability : Transient veya Transient Local

History : Keep last

3. Konfigürasyon Verisi

Uygulama tarafından kullanılan ve genellikle ilk açılışta okunan konfigürasyon verisi. Bu veriyi geç gelen katılımcıya vermek gerekir. Konfigürasyon verisini diskte saklamaya gerek vardır.

Reliability : Reliable. 

Durability : Persistent

History : Keep last

4. Event/Command Verisi

Mesela bir sistemi aç/kapa tarzı bir komut için VolatileDurability kullanılabilir., çünkü komut gönderildiği an için geçerlidir. Sonradan gelen birisinin bu komuta ihtiyacı yoktur.  

Reliability : Reliable. 

Durability : Volatile

History : Keep all veya Keep Last 1 veya Keep Last N

DDS Deadline Qos ve Liveliness Qos

Deadline Qos - Sistemin Yavaşladığını Anlamak İçindir
Deadline Qos belli aralıklarla güncelleme yapılacağını belirtiyor. Deadline süreleri uyumlu olan Writer ve Reader'lar birbirleri ile konuşabilir.

 Eğer Writer veya Reader belirtilen süre içinde güncellemede bulunmazsa her uygulama Listener vasıtasıyla haberdar ediliyor.

Bu durumda örneğin aynı veriyi sağlayan yedek kaynağa geçiş yapılabilir.

Deadline ve Liveliness sanki aynı şeylermiş gibi görünebilir ancak Liveliness sadece karşıdaki DataWriter'ın hayatta olduğunu belirtir. 

DataWriter hayatta olmasına rağmen halen Deadline süresini kaçırabilir, çünkü bir döngüde takılmıştır, sistemi yavaşlamıştır vs. 

Liveliness Qos - Sistemin Hayatta Olup Olmadığını Anlamak İçindir

Açıklaması şöyle.

In terms of applications, liveness means that the state of the application is healthy or not. If liveness is broken that means that the application itself is broken and cannot recover.

Bu Qos ile DataWriter halen ayakta olduğunu DataReader'lara bildiriyor. Eğer DataWriter bu Qos parametresindeki süreyi kaçırırsa, DataReader karşıdaki birimin artık ayakta olmadığını anlayabilir.

DDS DataReader Arayüzü

Giriş

DataReader şeklen şöyle

read metodu
Bir dizi okur. Şöyle yaparız.

dds::Topic<TempSensorType> tsTopic("TempSensorTopic");
dds::DataReader<TempSensorType> dr(tsTopic);
dds::SampleInfoSeq info;
TempSensorSeq data;
while (true) 
{
  dr.read(data, info);
  for (inti =0; i < data.length(); ++i)
    std::cout << data[i] << std::endl;sleep(1);
}

DDS DataWriter Arayüzü

Giriş

Açıklaması şöyle

DDS specification says that default value of Reliability for DataWriter is RELIABLE and for DataReader is BEST_EFFORT. 

...

They only way to get reliable communication going would be to modify the DataReader QoS to use RELIABLE reliability.

Açıklaması şöyle

Why does my DDS DataReader miss the first few samples?

Discovery is not an instantaneous event. It takes some time for the discovery process between applications to complete. The DDS DataWriter and DDS DataReader must discover each other before they can start communicating. Therefore, if you send data immediately after creating the RTI Connext DDS entities, DataReaders will not receive the first few samples because they are still in-process of discovering the DataWriters and vice versa.  This is true even when the DataWriters and DataReaders are reliable, because the Reliability QoS on its own does not guarantee delivery to DataReaders that have not been discovered, yet. This is expected behavior.

write metodu
Şöyle yaparız.

dds::domain::DomainParticipant dp(0);
dds::topic::Topic<MyType> topic(dp, "MyTopic");
dds::pub::Publisher pub(dp);
dds::pub::DataWriter<MyType> dw(pub, topic);

MyType t;
dw.write(t);

DDS DataReaderListener Arayüzü

Giriş

DataReader ile ilişkili bir sınıftır

on_requested_deadline_missed metodu
Örnek ver

on_requested_incompatible_qos metodu
Örnek ver

on_sample_rejected metodu
Örnek ver

on_liveliness_changed metodu
Örnek ver

on_data_available metodu
Genellikle sadece bu metod kullanılır. Gelen veriyi okuruz. Bazen gelen veriyi bir başka Executor içine atarak işlemek daha iyi olabiliyor.

Örnek

Şöyle yaparız. Burada DataReader benim istediğim tipte.

void on_data_available(DataReader<Foo>& the_reader) 
{
  ...
}

Örnek - C++
Şöyle yaparız. Parametre olarak geçilen DataReader pointer nesnesi benim istediğim DataReader tipine cast edilir. Daha sonra DataReader nesnesinin take_next_sample metodu çağrılır ve veri okunur.

Burada reader'dan veri kopyalanarak okunuyor.

void MyReader::on_data_available (DDS::DataReader* the_reader)
{
  MyReader * pReader = dynamic_cast<MyReader*> (the_reader);

  MyDataType data;
  SampleInfo sampleInfo;

  DDS::ReturnCode_t ret  = pReader->take_next_sample (data,sampleInfo);

  if (ret == DDS::RETCODE_OK && sampleInfo.valid_data == DDS_TRUE)
  {
    ...
  }
}

Örnek - Java
Şöyle yaparız. Burada Reader'dan okunan veri return_loan() ile tekrar iade ediliyor.

public void on_data_available(DataReader reader) {
  HelloWorldDataReader myReader = (HelloWorldDataReader)reader;

  try {
    myReader.take(_dataSeq, _infoSeq,
      ResourceLimitsQosPolicy.LENGTH_UNLIMITED,
      SampleStateKind.ANY_SAMPLE_STATE,
      ViewStateKind.ANY_VIEW_STATE,
      InstanceStateKind.ANY_INSTANCE_STATE);

    for(int i = 0; i < _dataSeq.size(); ++i) {
      SampleInfo info = (SampleInfo)_infoSeq.get(i);

      if (info.valid_data) {
        System.out.println(((HelloWorld)_dataSeq.get(i)).toString("Received",0));

      }
    }
  } catch (RETCODE_NO_DATA noData) {
    // No data to process
  } finally {
    myReader.return_loan(_dataSeq, _infoSeq);
  }
}

on_subscription_matched metodu
Örnek ver

on_sample_lost metodu
Örnek ver