AMiRo-OS

/'

µRtWare is a lightweight publish/subscribe middleware for real-time

applications. It was developed as part of the software habitat for the

Autonomous Mini Robot [1] (AMiRo) but can be used for other purposes as well.

Copyright (C) 2018..2020  Thomas Schöpping et al.

This program is free software: you can redistribute it and/or modify

it under the terms of the GNU General Public License as published by

the Free Software Foundation, either version 3 of the License, or

(at your option) any later version.

This program is distributed in the hope that it will be useful,

but WITHOUT ANY WARRANTY; without even the implied warranty of

MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

GNU General Public License for more details.

You should have received a copy of the GNU General Public License

along with this program.  If not, see <http://www.gnu.org/licenses/>.

'/

/'### INTRO ##################################################################'/

@startuml

title **µRtWare**\nPublish-Subscribe System

!include ../functions.iuml

/'### ENTITIES ###############################################################'/

!startsub ENTITIES

/' Message type. '/

$structure("urt_message_t") {

    'Pointer to the next message in a list.

  + {field} next : urt_message_t*

    'Pointer to some arbitrary (reusable) payload object.

  + {field} payload : void*

    'Origin time of the message.

  + {field} originTime : urt_osTime_t

    'Counter of HRT subscribers that did not consume the message yet.

  + {field} numHrtConsumersLeft : unsigned int

  .. URT_CFG_PUBSUB_PROFILING == true ..

    'Counter of overall subscribers that did not consume the message yet.

  + {field} numConsumersLeft : unsigned int

  __

    'Initializes a urt_message_t object.

  + {method} urtMessageInit (message : urt_message_t*, payload : void*) : urt_status_t

}

$group("subscriber") {

  /' Base subscriber type. '/

  $structure("urt_basesubscriber_t") {

      'Pointer to the topic, this subscriber subscribed to.

    + {field} topic : urt_topic_t*

      'Event listener to notify the node about new messages.

    + {field} evtListener : urt_osEventListener_t

      'Pointer to the message consumed most recently.

    + {field} lastMessage : urt_message_t*

      'Copy of the origin time of the message consumed most recently.

    + {field} lastMessageTime : urt_osTime_t

    .. URT_CFG_PUBSUB_PROFILING == true ..

      'Sum of all latencies.

    + {field} sumLatencies : uint64_t

      'Number of messages received.

    + {field} numMessagesReceived : uint64_t

  }

  /' Function type to be called when calculating the usefulness of a message. '/

  $function("urt_srtusefulnessfunc_t") {

    'Takes a delay and optionally parameters as arguments and returns a float in [0, 1].

    urt_srtusefulnessfunc_t (dt : urt_delay_t, params : void*) : float

  }

  /' NRT subscriber type. '/

  $structure("urt_nrtsubscriber_t") {

    .. URT_CFG_PUBSUB_PROFILING == true ..

      'Minimum latency ever detected.

    + {field} minLatency : urt_delay_t

      'Maximum latency ever detected.

    + {field} maxLatency : urt_delay_t

    __

      'Initializes a urt_nrtsubscriber_t object.

    + {method} urtNrtSubscriberInit (subscriber : urt_nrtsubscriber_t*) : urt_status_t

      'Tries to subscribe to a topic and optionally contributes a list of messages to the topic.

    + {method} urtNrtSubscriberSubscribe (subscriber : urt_nrtsubscriber_t*, topic : urt_topic_t*, messages : urt_message_t*) : urt_status_t

      'Fetches the next message in the buffer, optionally copies the payload and optionally returns the latency.

    + {method} urtNrtSubscriberFetchNextMessage (subscriber : urt_nrtsubscriber_t*, payload : void*, bytes : size_t, latency : urt_delay_t*) : urt_status_t

      'Fetches the latest message, optionally copies the payload and optionally returns the latency.

    + {method} urtNrtSubscriberFetchLatestMessage (subscriber : urt_nrtsubscriber_t*, payload : void*, bytes : size_t, latency : urt_delay_t*) : urt_status_t

      'Unsubscribes from a topic.

    + {method} urtNrtSubscriberUnsubscribe (subscriber : urt_nrtsubscriber_t*) : urt_status_t

  }

  /' SRT subscriber type. '/

  $structure("urt_srtsubscriber_t") {

      'Callback to calculate usefulness of a message

    + {field} usefulnesscb : urt_srtusefulnessfunc_t*

      'Optional parameters for the usefulness callback function.

    + {field} cbparams : void*

    .. URT_CFG_PUBSUB_PROFILING == true ..

      'Minimum latency ever detected.

    + {field} minLatency : urt_delay_t

      'Maximum latency ever detected.

    + {field} maxLatency : urt_delay_t

    __

      'Initializes a urt_srtsubscriber_t object.

    + {method} urtSrtSubscriberInit (subscriber : urt_srtsubscriber_t*) : urt_status_t

      'Tries to subscribe to a topic, sets all parameters and optionally contributes a list of messages to the topic.

    + {method} urtSrtSubscriberSubscribe (subscriber : urt_srtsubscriber_t*, topic : urt_topic_t*, messages : urt_message_t*, usefulnesscb : urt_srtusefulnessfunc_t*, cbparams : void*) : urt_status_t

      'Fetches the next message in the buffer, optionally copies the payload and optionally returns the latency.

    + {method} urtSrtSubscriberFetchNextMessage (subscriber : urt_srtsubscriber_t*, payload : void*, bytes : size_t, latency : urt_delay_t*) : urt_status_t

      'Fetches the latest message, optionally copies the payload and optionally returns the latency.

    + {method} urtSrtSubscriberFetchLatestMessage (subscriber : urt_srtsubscriber_t*, payload : void*, bytes : size_t, latency : urt_delay_t*) : urt_status_t

      'Calculates the usefulness of a message after the specified delay.

    + {method} urtSrtSubscriberCalculateUsefulness (subscriber : urt_srtsubscriber_t*, latency : urt_delay_t) : float

      'Unsubscribes from a topic.

    + {method} urtSrtSubscriberUnsubscribe (subscriber : urt_srtsubscriber_t*) : urt_status_t

  }

  /' FRT subscriber type. '/

  $structure("urt_frtsubscriber_t") {

    .. URT_CFG_PUBSUB_QOS_DEADLINECHECKS == true ..

      'Maximum temporal offset between creation and consumption of messages.

    + {field} deadlineOffset : urt_delay_t

    .. URT_CFG_PUBSUB_QOS_JITTERCHECKS == true ..

      'Maximum expected jitter.

    + {field} maxJitter : urt_delay_t

    .. URT_CFG_PUBSUB_QOS_JITTERCHECKS == true ||  URT_CFG_PUBSUB_PROFILING == true ..

      'Minimum latency ever detected (to calculate jitter).

    + {field} minLatency : urt_delay_t

      'Maximum latency ever detected (to calculate jitter).

    + {field} maxLatency : urt_delay_t

    __

      'Initializes a urt_frtsubscriber_t object.

    + {method} urtFrtSubscriberInit (subscriber : urt_frtsubscriber_t*) : urt_status_t

      'Tries to subscribe to a topic, sets all parameters and optionally contributes a list of messages to the topic.

    + {method} urtFrtSubscriberSubscribe (subscriber : urt_frtsubscriber_t*, topic : urt_topic_t*, messages : urt_message_t*, deadline : urt_delay_t, jitter : urt_delay_t) : urt_status_t

      'Fetches the next message in the buffer, optionally copies the payload and optionally returns the latency.

    + {method} urtFrtSubscriberFetchNextMessage (subscriber : urt_frtsubscriber_t*, payload : void*, bytes : size_t, latency : urt_delay_t*) : urt_status_t

      'Fetches the latest message, optionally copies the payload and optionally returns the latency.

    + {method} urtFrtSubscriberFetchLatestMessage (subscriber : urt_frtsubscriber_t*, payload : void*, bytes : size_t, latency : urt_delay_t*) : urt_status_t

      'Checks whether a message is valid after the specified delay.

    + {method} urtFrtSubscriberCalculateValidity (subscriber : urt_frtsubscriber_t*, latency : urt_delay_t) : bool

      'Unsubscribes from a topic.

    + {method} urtFrtSubscriberUnsubscribe (subscriber : urt_frtsubscriber_t*) : urt_status_t

  }

  /' HRT subscriber type. '/

  $structure("urt_hrtsubscriber_t") {

      'Pointer to the next HRT subscriber in a list.

    + {field} next : urt_hrtsubscriber_t*

    .. URT_CFG_PUBSUB_QOS_DEADLINECHECKS == true ..

      'Maximum temporal offset between creation and consumption of messages.

    + {field} deadlineOffset : urt_delay_t

      'QoS Timer to detect missed deadlines.

    + {field} qosDeadlineTimer : urt_osTimer_t

    .. URT_CFG_PUBSUB_QOS_JITTERCHECKS == true ..

      'Maximum expected jitter.

    + {field} maxJitter : urt_delay_t

    .. URT_CFG_PUBSUB_QOS_JITTERCHECKS == true || URT_CFG_PUBSUB_PROFILING == true..

      'Minimum latency ever detected (to calculate jitter).

    + {field} minLatency : urt_delay_t

      'Maximum latency ever detected (to calculate jitter).

    + {field} maxLatency : urt_delay_t

    .. URT_CFG_PUBSUB_QOS_RATECHECKS == true ..

      'Expected rate at which data is published.

    + {field} expectedRate : urt_delay_t

    __

      'Initializes a urt_hrtsubscriber_t object.

    + {method} urtHrtSubscriberInit (subscriber : urt_hrtsubscriber_t*) : urt_status_t

      'Tries to subscribe to a topic, sets all parameters and optionally contributes a list of messages to the topic.

    + {method} urtHrtSubscriberSubscribe (subscriber : urt_hrtsubscriber_t*, topic : urt_topic_t*, messages : urt_message_t*, deadline : urt_delay_t, rate : urt_delay_t, jitter : urt_delay_t) : urt_status_t

      'Fetches the next message in the buffer, optionally copies the payload and optionally returns the latency.

    + {method} urtHrtSubscriberFetchNextMessage (subscriber : urt_hrtsubscriber_t*, payload : void*, bytes : size_t, latency : urt_delay_t*) : urt_status_t

      'Fetches the latest message, optionally copies the payload and optionally returns the latency.

    + {method} urtHrtSubscriberFetchLatestMessage (subscriber : urt_hrtsubscriber_t*, payload : void*, bytes : size_t, latency : urt_delay_t*) : urt_status_t

      'Unsubscribes from a topic.

    + {method} urtHrtSubscriberUnsubscribe (subscriber : urt_hrtsubscriber_t*) : urt_status_t

  }

} /'subscriber'/

/' Publisher type. '/

$structure("urt_publisher_t") {

    'Pointer to the topic for publishing.

  + {field} topic : urt_topic_t*

  .. URT_CFG_PUBSUB_PROFILING == true ..

    'Counter of attempts to publish a message.

  + {field} publishAttempts : uint64_t

    'Counter of failed attempts to publish a message.

  + {field} publishFails : uint64_t

  __

    'Initializes a urt_publisher_t object and contributes an optional list of messages.

  + {method} urtPublisherInit (publisher : urt_publisher_t*, topic : urt_topic_t*, messages : urt_message_t*) : urt_status_t

    'Publishes a message via the associated topic.

  + {method} urtPublisherPublish (publisher : urt_publisher_t*, payload : void*, n : size_t, t : urt_osTime_t, timeout : urt_delay_t) : urt_status_t

}

/' Topic type. '/

$structure("urt_topic_t") {

    'Pointer to the next topic in a list.

  + {field} next : urt_topic_t*

    'Identifier of the topic.

  + {field} id : urt_topicid_t

    'Mutex lock for exclusive access.

  + {field} lock : urt_osMutex_t

    'Event source to inform all subscribers when a new message is published.

  + {field} evtSource : urt_osEventSource_t

    'Number of HRT subscribers.

  + {field} numHrtSubscribers : unsigned int

    'List of HRT subscribers, orderd by their expected rate (most critical first).

  + {field} hrtSubscribers : urt_hrtsubscriber_t*

    'Condition variable to inform waiting publishers when a blocked message becomes available.

  + {field} hrtReleased : urt_osCondvar_t

    'Mandatory message, each Topic holds.

  + {field} mandatoryMessage : urt_message_t

    'Pointer to the latest message.

  + {field} latestMessage : urt_message_t*

  .. URT_CFG_PUBSUB_QOS_RATECHECKS == true ..

    'Timer to check for missed rates.

  + {field} qosRateTimer : urt_osTimer_t

  .. URT_CFG_PUBSUB_PROFILING == true ..

    'Variable to count how many (non-hrt) subscribers did not fetch a message before it was reused.

  + {field} numDiscardedMessages : uint64_t

    'Number of overall subscribers.

  + {field} numSubscribers : unsigned int

  __

    'Initializes an urt_topic_t object.

  + {method} urtTopicInit (topic : urt_topic_t*, id : urt_topicid_t) : urt_status_t

}

!endsub

/'### DEPENDENCIES & LAYOUT ##################################################'/

!startsub DEPENDENCIES

urt_message_t "1" o-- "0..1" urt_message_t

urt_nrtsubscriber_t --|> urt_basesubscriber_t

urt_nrtsubscriber_t ..> urt_message_t

urt_srtsubscriber_t --|> urt_basesubscriber_t

urt_srtsubscriber_t "1" o-- "0..1" urt_srtusefulnessfunc_t

urt_srtsubscriber_t ..> urt_message_t

urt_frtsubscriber_t --|> urt_basesubscriber_t

urt_frtsubscriber_t ..> urt_message_t

urt_hrtsubscriber_t --|> urt_basesubscriber_t

urt_hrtsubscriber_t "1" o-- "0..1" urt_hrtsubscriber_t

urt_hrtsubscriber_t ..> urt_message_t

urt_publisher_t "1" o- "1" urt_topic_t

urt_publisher_t ..> urt_message_t

urt_topic_t "1" o-- "0..1" urt_topic_t

urt_topic_t "1" o- "0..1" urt_hrtsubscriber_t

urt_topic_t "1" *-- "1" urt_message_t

!endsub

/'### OUTRO ##################################################################'/

@enduml