Kyoungho An
email: kyoungho.an@gmail.com

Shweta Khare
email: shweta.p.khare@vanderbilt.edu

Akram Hakiri
email: akram.hakiri@gmail.com

Aniruddha Gokhale
email: a.gokhale@vanderbilt.edu

An Autonomous and Dynamic Coordination and Discovery Service for Wide-Area Peer-to-peer Publish/Subscribe

Keywords: Computer systems organization →Cloud computing, Peerto-peer architectures, •Applied computing →Event-driven architectures, Data Distribution Service, Pub/Sub, Discovery, Coordination

come

INTRODUCTION

Emerging paradigms, such as the Internet of ings (IoT), connect machines and devices in a loosely coupled manner to form intelligent and large-scale systems. A class of IoT, referred to as Industrial IoT (IIoT) [START_REF] Annunziata | Industrial Internet: Pushing the Boundaries of Minds and Machines[END_REF], found in domains such as transportation, healthcare, manufacturing, and energy requires di erent quality of service (QoS) properties, such as timeliness, reliability, and security, for their applications that are geographically distributed and operate over wide area networks (WANs). For example, wind farms have di erent requirements for data analysis depending on the type of analysis (e.g., frequency of data arrival for machine-level analysis is every 40 milliseconds and for plant-level analysis is every one second). e key to successful data analysis relies on how e ective is the system in collecting and delivering data across a large number of entities at Internet-scale in a timely, and reliable manner. 1e publish/subscribe (pub/sub) communication paradigm is a ractive for these emerging systems since it provides a scalable and decoupled data delivery mechanism between communicating peers. Speci cally, the need to scalably and reliably disseminate large volumes of IIoT data in real-time motivates the use of scalable, datacentric pub/sub with support for con gurable QoS properties that can operate over WANs. Many pub/sub messaging solutions exist today that can operate over multiple networks including industrial solutions [START_REF] Hintjens | ZeroMQ: Messaging for Many Applications[END_REF][START_REF][END_REF][START_REF] Vinoski | Advanced message queuing protocol[END_REF] and research e orts [START_REF] Carzaniga | Design and evaluation of a wide-area event noti cation service[END_REF][START_REF] Esposito | Reliable publish/subscribe middleware for time-sensitive internet-scale applications[END_REF][START_REF] Peter | Hermes: A distributed event-based middleware architecture[END_REF]. Some of these even support QoS properties, such as availability [START_REF] Esposito | Reliable publish/subscribe middleware for time-sensitive internet-scale applications[END_REF][START_REF] Peter | Hermes: A distributed event-based middleware architecture[END_REF], con gurable reliability [START_REF][END_REF], durability [START_REF] Hintjens | ZeroMQ: Messaging for Many Applications[END_REF], and timeliness [START_REF] Carvalho | Scalable QoS-based event routing in publish-subscribe systems[END_REF][START_REF] Kim | E cacy of techniques for responsiveness in a widearea publish/subscribe system[END_REF]. However, these solutions either do not address the scalable datacentric needs, or tend to support only a subset of properties at a time and in most cases, support for con gurable QoS properties, security, and resilience is lacking.

e Object Management Group (OMG)'s Data Distribution Service (DDS) [START_REF] Omg | Data Distribution Service speci cation[END_REF] standard for data-centric pub/sub holds substantial promise for IIoT applications because of its support for con gurable QoS policies, dynamic discovery of end points, low-latency and resilience due to its peer-to-peer architecture. However, there still remain a number of unresolved challenges in using OMG DDS in IIoT systems deployed over multiple, distributed networks. For instance, DDS uses multicast as a default transport to automatically discover peers in a system. If the endpoints are located in isolated networks that do not support multicast, then these endpoints cannot be discovered by each other. Secondly, even if these endpoints were discoverable, because of network rewalls and network address translation (NAT), peers may not be able to deliver data to the destination endpoints.

One approach to supporting OMG DDS over multiple distributed networks relies on broker-based solutions [START_REF] Hakiri | Supporting End-to-end Scalability and Real-time Event Dissemination in the OMG Data Distribution Service over Wide Area Networks[END_REF][START_REF] Jose | A content-aware bridging service for publish/subscribe environments[END_REF], where brokers are used to inter-connect peers in di erent networks. It is thus conceivable to think that these broker-based solutions in conjunction with the data-centric and con gurable QoS features provided by OMG DDS can readily provide a solution for IIoT. However, it is still challenging to realize such capabilities when the deployed system spans a large number of networks comprising pub/sub endpoints that illustrate heterogeneity in terms of topics, their types, and requested versus o ered QoS policies. Additionally, such solutions tend to lack autonomous and dynamic discovery and coordination of brokers to connect pub/sub peers in multiple, disparate networks on demand.

Further, as industrial systems progressively integrate an increasing number of sub-systems located in multiple disparate networks, the number of deployed brokers may become very large. Consequently, the amount of e ort to manage these dispersed brokers becomes unwieldy due to the distributed state and con guration. Moreover, forming an e cient overlay network of brokers that o ers both scalability and low latency becomes even harder.

To address these technical concerns while still bene ing from technologies, such as OMG DDS, and to realize a practical and readily deployable solution that IIoT applications can leverage, we present PubSubCoord, which provides an autonomous and dynamic coordination and discovery service for geographically distributed brokers to transparently connect pub/sub endpoints and realize scalable and low-latency, data-centric pub/sub systems. Our earlier work presented only a vision of PubSubCoord [START_REF] An | PubSubCoord: A Cloud-enabled Coordination Service for Internet Scale OMG DDS Applications[END_REF][START_REF] Kyoungho An | Wide Area Network-scale Discovery and Data Dissemination in Data-centric Publish/Subscribe Systems[END_REF]. is paper signi cantly extends this vision by delving into the technical details of its design, and makes the following contributions:

• To address the scalability and low latency requirements of data dissemination across multiple distributed networks, and to limit the complexity of managing a broker network, PubSubCoord introduces a two-level broker hierarchy deployed over a pub/sub overlay network, which in turn incurs only two hops in the worst case in the data dissemination path across the overlay of distributed, isolated networks. • To achieve autonomous and dynamic discovery and data routing between brokers, PubSubCoord exploits and extends the open-source ZooKeeper coordination service [START_REF] Hunt | ZooKeeper: wait-free coordination for internet-scale systems[END_REF] in a novel way to create dissemination paths for the dynamic overlay network of brokers and endpoints. • For those dissemination paths that need both low latency and high-availability requirements, PubSubCoord trades o resource usage in favor of deadline-aware overlays that build multiple, redundant paths between brokers. • An implementation of PubSubCoord using Real Time Innovation (RTI)'s implementation of OMG DDS as the underlying pub/sub messaging system is presented and empirically evaluated to demonstrate the feasibility of the solution and its ability to support end-to-end QoS properties without incurring any undue overhead introduced by the broker architecture.

e remainder of this paper is organized as follows: Section 2 describes the design and implementation of PubSubCoord; Section 3 presents experimental results validating our claims; Section 4 compares PubSubCoord with related work; and Section 5 presents concluding remarks and alludes to lessons learned and future work.

To make the paper self-contained, Appendix A provides background information on the underlying technologies used in PubSubCoord.

DESIGN AND IMPLEMENTATION OF PUBSUBCOORD

In the realm of (I)IoT, there is an emerging trend towards exploiting the entire spectrum of resources ranging from the edge to the cloud [START_REF] Bonomi | Fog computing and its role in the internet of things[END_REF][START_REF] Satyanarayanan | Mobile Computing: e Next Decade[END_REF]. e separation of concerns and latency/scalability bene ts of edge-cloud architectures motivates the two levels of the broker hierarchy in our PubSubCoord design. Figure 1 shows the PubSubCoord architecture depicting three layers: a coordination layer, a pub/sub overlay layer, and the physical network layer. e pub/sub overlay comprises the two-level overlay network of brokers and pub/sub endpoints in a system. An edge broker is directly connected to endpoints in a local area network (LAN) to serve as a gateway to other endpoints placed in di erent networks. A routing broker serves as a mediator to route data between edge brokers according to assigned and matched topics that are present in the global data space spread across the WANs. e coordination layer comprises an ensemble of ZooKeeper servers used for coordination between the brokers. ZooKeeper has been used by many industrial solutions, such as FaRM [START_REF] Dragojević | FaRM: Fast remote memory[END_REF], Ambry [START_REF] Shadi A Noghabi | Ambry: LinkedIn's Scalable Geo-Distributed Object Store[END_REF], CoRAL [START_REF] Vora | CoRAL: Con ned Recovery in Distributed Asynchronous Graph Processing[END_REF] and Net ix's Curator for distributed coordination. e routing broker layer resides in the cloud (or fog) along with the coordination logic while the edge broker layer resides in the edge.

Data dissemination in PubSubCoord can be explained using an example from Figure 1. Pi{T } denotes a publisher i that publishes topic T (similarly for a subscriber S). Since there are no endpoints interested in topic A other than publisher P1 and subscriber S1, they communicate only within the local network A via either UDP-based multicast or unicast for low latency without incurring a hop to the routing broker layer. P2, P4, and S2 are interested in topic B but are deployed in di erent networks. So their communications are routed through a routing broker that is responsible for topic B.

In the remainder of this section, we discuss the di erent design decisions and implementation aspects of PubSubCoord.

Scalability via Separation of Concerns

Context and challenges: Traditional WAN-based pub/sub systems tend to form an overlay network of brokers to which endpoints can be connected. e brokers exchange subscriptions they receive from subscribers that are used to build routing paths from publishers. e main challenge in this approach stems from having to build routing states among brokers to route data e ciently according to matching subscriptions. Maintaining such distributed state may become unwieldy. Secondly, in traditional broker-based pub/sub systems, if some broker were to fail, it halts not only the pub/sub service for endpoints connected to this broker but also service for endpoints connected to other brokers that use this failed broker as an intermediate routing broker. Moreover, the dissemination latency su ers due to multiple broker hops.

Solution approach: To resolve these challenges, PubSubCoord's broker overlay layer is structured as just a two-tier architecture of brokers with strict separation of responsibilities: at the bo om tier are edge brokers that manage pub/sub issues within an isolated network, and at the top tier are routing brokers in the cloud, which route tra c among di erent edge brokers. us, we incur a maximum two-hop latency for communication across isolated LANs.

Design details, consequences and their resolution: Our solution clusters edge brokers by matching topics and routes data through routing brokers. Each routing broker is responsible for handling only a certain number of topics so as to balance the topic load and number of connections between edge brokers.

An immediate consequence of our design decision is having to decide how many routing brokers to maintain, how many topics to be handled by each routing broker, and how to organize them in the system. Having only one routing broker would be problematic since it cannot scale to handle the substantial routing load stemming from the dissemination of multiple di erent topic data among the large number of endpoints. Having multiple routing brokers and organizing them in multiple levels of hierarchy similar to domain name service (DNS) would not be acceptable either since it would complicate the management of topics and recovery from failures because the routing state and topic management would get distributed across multiple levels. Secondly, multiple levels as in DNS introduces multiple routing hops, which will impact latency of distribution and thereby scalability of the system. For that reason, we maintain a at tier of routing brokers.

e number of routing brokers and topics managed by each routing broker is determined by the end-to-end performance requirements of the pub/sub ows. us, a solution that can elastically scale the number of routing brokers and balance the number of topics handled by each broker is needed. For that reason, the routing broker tier is placed as a cluster in the cloud where resources can be elastically scaled up/down depending on the demand. e number of routing brokers can thus be scaled up/down to dynamically adapt to system load using cloud-based autoscaling algorithms, such as the one we proposed in prior work [START_REF] Roy | E cient Autoscaling in the Cloud using Predictive Models for Workload Forecasting[END_REF]. us, our design enables dynamic adaptation to system load and autoscaling to existing demand. is design can also take care of faults in routing brokers where a new routing broker can be spawned on demand in case of failure and the failed broker's topics can be re-distributed among existing brokers.

Low Latency for Intra-LAN Data Dissemination

Context and challenges: Systems that cater to supporting WANscale operations may tend to overlook LAN-speci c issues and force a one-size-ts-all WAN-scale solution even for a LAN-scale operation, thereby incurring unnecessary overhead and performance penalties on applications. For PubSubCoord this implies that any intra-LAN data dissemination should not incur the overhead of WAN latencies stemming from having to utilize the cloud-based routing broker layer. Solution approach: It is in this regard that the two-level broker solution of PubSubCoord and separation of concerns addresses these potential problems, where the edge broker layer handles all the LAN-speci c pub/sub issues.

Design details, consequences and their resolution: In our solution, any pub/sub tra c that is local (i.e., where publishers and subscribers reside in the same isolated network) is not allowed to reach the routing brokers; rather the tra c and dissemination is handled by the edge brokers themselves thereby avoiding the penalty of a round-trip WAN latency and in turn avoiding undue overhead imposed on the routing brokers.

A consequence of our design is that the edge broker may get overloaded and/or fail. To that end, elastic solutions and load balancing decisions similar to that used in the cloud can be used for edge brokers. In fact, the emerging edge-cloud architectures are already supporting elastic solutions across the spectrum of edge-cloud resources [START_REF] Byung-Gon Chun | Clonecloud: elastic execution between mobile device and cloud[END_REF]. Moreover, faults can be handled using a variety of mechanisms including leveraging the range of fault tolerance solutions we have developed in prior work [8, 10-12, 22, 24, 44].

WAN-scale Autonomous and Dynamic Endpoint Discovery and Dissemination

Context and challenges: In IIoT systems, the publishers and subscribers that are matched with each other are most likely distributed across separate isolated networks, and moreover they may join or leave the system dynamically. us, publishers and subscribers must be able to dynamically discover each other and a dissemination route needs to be established between the communicating entities. As discussed earlier, the dynamic discovery mechanisms of OMG DDS do not work at WAN scale due to limitations with IP multicast in the WAN and NAT/ rewall issues.

Solution approach: To address this need, PubSubCoord uses a coordination layer (top layer shown in Figure 1) comprising an ensemble of ZooKeeper [START_REF] Hunt | ZooKeeper: wait-free coordination for internet-scale systems[END_REF] servers, which help brokers discover each other and build broker overlay networks using the PubSubCoord coordination logic. ZooKeeper is an open-source, highly e cient, centralized and replicated service which provides generic distributed coordination and synchronization primitives such as leader election, barrier synchronization,locking, group membership,sharing con guration metadata, etc. e data model of ZooKeeper is structured like a le system in the form of znodes (i.e., a ZooKeeper data object containing its path and data content). Applications using the znodes are required to specialize the znodes to add applicationspeci c semantics. ZooKeeper provides a watch mechanism to notify a client of ZooKeeper of a change to a znode that is being watched by that client. Specializing the ZooKeeper znode structure for our needs and using its watch mechanisms provides us with a capability that can take autonomous and dynamic decisions.

Design details, consequences and their resolution: e child znodes for every unique topic represent the endpoints, i.e., publishers and subscribers, associated with it. e leader znode is used to elect a leader among the routing brokers. A leader routing broker makes load balancing decisions for topic distribution among the cluster of routing brokers. e broker znode has child znodes for each routing broker where its location information (i.e., IP address and port number of a routing broker) is stored.

e leader uses this information to associate a selected routing broker's location to a topic znode a er topic assignment.

Dynamic changes in the system (e.g., broker or pub/sub endpoint join/leave, topic creation/deletion) are re ected in the creation, update, and deletion of znodes in the znode-tree. Brokers use the watch mechanism to set watches on interesting znodes to receive change noti cations and to take actions autonomously. PubSub-Coord exploits the ephemeral mode feature of ZooKeeper, where a speci c znode in the tree (and its subtree) is automatically deleted from the tree when the client session handling this znode is lost thereby providing automatic cleanup of state.

We rely on ZooKeeper to provide the desired fault tolerance and persistence of the znode tree. Details on all the interactions that take place in this context are provided in Section 2.6 where we show how brokers discover each other and establish communication routes.

Overload Management and Fault Tolerance

Context and challenges: Performance of PubSubCoord can be impacted by at least two factors: load and failures in the routing and edge brokers; and network congestion on the two hop route between edge and routing brokers over the broker overlay.2 Load on a routing broker depends on the number of topics it manages and the number of edge brokers it interconnects through itself. us, appropriate strategies are needed for routing broker load balancing. In the case of faults, although many kinds of faults are possible, we focus only on tolerating failures in the routing broker layer.

Routing Broker Load Management

. Solution approach: Load balancing is handled by the leader routing broker, which is elected among the routing brokers using ZooKeeper. In the current implementation, the leader routing broker assigns a new topic to the least loaded routing broker in terms of number of topics.

Design details, consequences and their resolution: To elect a leader in a consistent and safe manner,PubSubCoord uses ZooKeeper's leader znode for routing brokers to write themselves on the znode so as to be elected as a leader (i.e., voting process). e routing broker that gets to write rst becomes a leader since the znode is locked therea er (i.e., no one can write on the znode unless the leader fails). e remaining routing brokers become workers. Worker routing brokers relay pub/sub data between edge brokers.

e leader routing broker can also serve as a worker routing broker.

A leader routing broker must manage the cluster of routing brokers and assign topics to workers in a way that balances the load. It does this by selecting the least loaded worker, which currently is decided based on the number of adopted topics by that worker. By using the Strategy design pa ern our design can allow plugging in other load balancing schemes (e.g., least loaded based on CPU utilization, the number of connections, QoS policies, or DHT-based load balancing).

Deadline-aware Overlay Optimizations . Solution approach:

e second cause of performance bo lenecks stems from the congested two-hop route connecting edge brokers via a routing broker. To overcome this problem, PubSubCoord also supports an optimization to both improve reliability and latency by providing an additional one hop path over the overlay that directly connects communicating edge brokers.

Design details, consequences and their resolution: Figure 3 illustrates the idea. ese optimizations can be leveraged by pub/sub ows that require stringent assurances on reliable and deadlinedriven data delivery. For example, OMG DDS uses deadline QoS as a contract between pub/sub ows, which can be used to express the maximum duration of a sample to be updated. For those event streams requiring strict deadlines, multi-path overlay networks build an alternative, additional path directly between edge brokers thereby reducing the number of hops to just one. Note that this strategy is vulnerable to abuse by applications, and doing this for every edge broker will be infeasible due to the very large connection state that every edge broker must manage. us, higher level policies and admission control mechanisms will be needed to restrict such optimizations to the most critical ows.

Broker Fault Tolerance

. Solution approach: PubSubCoord o ers tolerance to routing broker failures, which can be of two kinds: worker failure and leader failure. When the worker fails, the leader reassigns topics handled by that failed broker to another worker routing broker to avoid service cessation. If the load is too high, the cloud will elastically scale the number of routing brokers. If a leader fails, the routing brokers vote for another leader again. On (re)assignment or failure of routing broker, PubSubCoord leverages ZooKeeper's watch mechanism to notify the appropriate edge brokers to update their paths to the right routing broker.

Design details, consequences and their resolution: ere may be transient periods of time when brokers fail and their load has to be migrated. We have not yet evaluated the impact of such transient unavailability of service and its impact on loss of samples being disseminated. Our ongoing work is designing solutions to minimize the impact of these situations. Second, the ZooKeeper server itself may fail. To provide a scalable and fault-tolerant service at the coordination layer, ZooKeeper supports the notion of an ensemble, and a leader of the ensemble synchronizes data between distributed servers to provide consistent coordination events to clients (i.e., brokers in our solution) and avoid single points of failure.

Concrete Instantiation of PubSubCoord with OMG DDS

To demonstrate our ideas and enable a readily deployable solution, we have implemented PubSubCoord in the context of OMG Data Distribution Service (DDS) where the pub/sub endpoints use the OMG DDS technology, speci cally the RTI Connext DDS implementation, along with its requested-o ered QoS model. We have used Curator, 3 which is a high-level API that simpli es using ZooKeeper, and provides useful recipes such as leader election and caches of

PubSubCoord with OMG DDS in Action

So far we have presented the architectural elements and implementation details of PubSubCoord. We now present details on the runtime interactions among the architectural elements that realizes the various capabilities of PubSubCoord. We describe how the brokers interact and the algorithms they execute to update their internal states used in routing the streamed pub/sub data. We show these interactions concretely in the context of publishers and subscribers that use OMG DDS and its QoS policies.

2.6.1 Routing Broker Responsibilities. Figure 4 illustrates the sequence diagram showing the runtime interactions of the routing brokers.

e corresponding algorithm executed by the routing broker is captured in Algorithm 1. is algorithm is predominantly event-driven, i.e., it is made up of callback functions that are invoked when some condition is satis ed.

ese callback functions are invoked by ZooKeeper due to the di erent watch conditions. e Routing Service shown in the gure and used in the algorithm are the capabilities at the edge broker that bridge the isolated network to the outside world. In our case, it is supplied by RTI's DDS implementation. e algorithm and sequence of steps for the routing broker operate as follows: Each routing broker initially connects to the ZooKeeper leader as a client of the ZooKeeper service. e cluster of routing brokers subsequently elect a leader among themselves using the process described in Section 2.3 that uses the leader znode. Once a leader is elected, it registers a listener (i.e., event detector that is noti ed when the registered znode changes) on the topics znode (shown in Figure 2) to receive topic relevant events (e.g., creation or deletion of topics). e following callback functions are implemented by the routing brokers:

Leader

• -is function is invoked when a znode for a worker routing broker is updated with an assigned topic by a leader routing broker.

•

-is function is invoked when children pub/sub endpoints of a znode for an assigned topic are created, deleted, or updated.

Every worker routing broker registers a listener on the znode for itself to receive topic assignment events updated by a leader routing broker. In the callback function, the znode for the routing broker stores a set of topics. When the topic set is updated by the leader (e.g., the leader assigns a new topic to the worker routing broker), it applies the changes by creating a cache for the assigned topic and its listener to receive events relevant to endpoints interested in the assigned topic.

When an endpoint is created or deleted in an isolated network, their edge brokers create or delete znodes for endpoints and these events will trigger the function in the routing brokers that are responsible for the topics involved with the endpoints. e metadata of the znode cache for an endpoint (ep in the callback function) contains the locator of an edge broker where the endpoint is located as well as the topic name, type, and QoS se ings.

If the event type is creation, it adds the locator of the edge broker to the DDS Routing Service running in the routing broker if it does not exist. erea er, it requests the DDS Routing Service to create a route for the topic based on the information provided by the content of the ep znode from this routing broker to the edge broker, if it does not exist. If the event type is deletion, it has to delete the locator and the topic route from the DDS Routing Service on the condition that no endpoints for that topic still exist.

Consider a concrete example. Using Figure 4, when TopicA is created, the leader routing broker assigns the topic to the least loaded worker, which currently is decided based on the number of adopted topics by that worker. Next, the leader updates a locator of the assigned worker broker on the corresponding znode that is created for TopicA, i.e., a child of topics znode -see the le most node in row three of Figure 2. is locator information will then be used by edge brokers interested in TopicA.

A worker routing broker initially registers listeners on a znode for itself (i.e., a child of broker znodes) to receive topic assignment events, which occur when the assigned topics znode is updated by a leader routing broker. When the worker routing broker is informed that it must handle a speci c topic, such as TopicA, it then registers a listener on pub/sub znodes for that particular assigned topic (e.g., children of topic A znode) to receive endpoint discovery events, such as creation of publisher or subscriber endpoints interested in TopicA. When an endpoint for TopicA is created and the worker routing broker is noti ed, it establishes data dissemination paths to edge brokers. For this data dissemination, PubSubCoord relies on the underlying pub/sub messaging systems' broker capabilities.

2.6.2 Edge Broker Responsibilities. Figure 5 shows the corresponding sequence diagram for edge brokers, and Algorithm 2 describes the logic of the callback functions implemented by the edge broker. • : is function is invoked when a topic znode managed by an edge broker is updated with a locator of an assigned worker routing broker.

• : is function is invoked when an endpoint in a network is deleted. e edge broker operates as follows: Like routing brokers, edge brokers initially connect to the ZooKeeper servers as clients of the ZooKeeper service. In the context of OMG DDS, edge brokers make use of built-in entities (i.e., special pub/sub entities for discovering peers and endpoints in a network supported by OMG DDS) to discover endpoints in local networks. For example, when a pub or sub endpoint interested in TopicA is created within some isolated network, the built-in entities receive discovery events via multicast, and then edge brokers create znodes for the created endpoints.

Edge brokers register a listener on a topic znode (e.g., topic A in Figure 2) in which the created endpoint is interested in to obtain the locator of the routing broker that is in charge of that particular topic. Once a locator of a routing broker is obtained, an edge broker initiates a data dissemination path to the routing broker through the routing capabilities that are assumed to be collocated with the edge broker. e callback function rst creates a znode for a created endpoint (i.e. ep in Algorithms 2) that contains the topic name, type, and QoS se ings. If a relevant topic to the created endpoint has not appeared in an edge broker before, a cache for the topic znode and its listener for the topic are created to receive locator information of an assigned worker routing broker. When the znode for the topic is updated by a leader routing broker, it triggers the callback described in Algorithm 2. In the callback function, each topic znode stores the locator of the worker routing broker that is responsible for the topic. e locator of a routing broker is added to the routing capability of the edge broker to establish a communication path between the edge broker and a worker routing broker. e callback function deletes the znode for the existing endpoint, and deletes it from the multi-set for topics. Next, it checks if the multi-set contains the topic of the deleted endpoint. If the topic is contained in the multi-set, it means other endpoints are still interested in the topic. If it is empty, it means no other endpoints that are interested in the topic exists, and that the cache and its listener need to be removed. e multi-set data structure for topics is used because there may still exist endpoints interested in topics relevant to deleted endpoints.

To support mobility or termination of endpoints, PubSubCoord relies on a timeout event that occurs by virtue of using the DDS liveliness QoS policy (which is used to detect disconnected endpoints where the timeout values are con gurable) and accordingly the znodes (which operate in the ephemeral mode) for those endpoints are deleted from the coordination servers and the route maintained at the edge broker is also terminated.

EXPERIMENTAL VALIDATION OF PUBSUBCOORD

is section presents the experimental results we conducted to evaluate the latency bene ts, scalability and deadline-aware overlays of PubSubCoord.

Overview of Testbed Con gurations and Testing Methodology

Our testbed is a private cloud managed by OpenStack comprising 60 physical machines each with 12 cores and 32 GB of memory. Both Edge and Routing broker instances run in their own Virtual Machine (VM), while multiple publisher and subscriber test applications share a VM. Each Virtual Machine (VM) used in our experiment was con gured with one virtual CPU and 2 GB RAM. To experiment in a multi-network environment, we used Neutron4 , an OpenStack project for networking as a service, to create virtual networks. All publish/subscribe end-points, were implemented using RTI Connext 5.15 and were con gured to use reliability QoS to avoid data loss at the transport level through data retransmission;

history QoS to keep all historical data; and durability QoS to make it possible for late-joining subscribers to obtain previously published samples. e lifespan QoS is set to 60 seconds so publishers guarantee persistence for 60 seconds. Depending on the application's requirements, the QoS policies can be varied and hence the performance results may change according to di erent QoS se ings.

Each publisher test application sends a 64 byte data sample every 50 milliseconds (other sizes could be used). We used netem6 network emulator to emulate 20 milliseconds roundtrip LAN latencies and 80 milliseconds roundtrip WAN latencies. We measure the end-toend latency from publishers to subscribers to evaluate our solution. End-to-end latency was calculated as the time di erence between the send timestamp at the publisher and reception timestamp at the subscriber.

e Precise Time Protocol (PTP) [START_REF] Carroll | IEEE 1588 Precision Time Protocol (PTP)[END_REF] was used for ne-grained time synchronization across all VMs. We collect latency values of 5,000 samples in total for each subscriber and use values only a er 1,000 samples since the latency values of initial samples are not consistent due to coordination and discovery overhead (e.g., time for discovery of brokers and creation of routes).

e broker CPU usage is also measured along with the latency values to understand how di erent se ings, i.e., number of topics per network and number of routing brokers, a ect dissemination scalability.

Data Locality Results

Edge Broker layer in PubSubCoord is responsible for dissemination of local tra c thereby preventing WAN latencies in sending data all the way to a cloud back-end for local data dissemination. To evaluate the latency bene ts of the Edge Broker layer, we measure the end-to-end latency of data dissemination under di erent values of data locality, i.e, fraction of topics in an isolated network which are local to the network and do not have interested subscribers in another network which would necessitate going through the cloud Routing Broker layer.

We created ve local networks each serviced by its own Edge Broker VM and ve Routing Broker VMs. In each of the ve isolated networks we used upto 10 client VMs for hosting pub/sub test applications where each client VM can host a maximum of 20 pub/sub test applications. We created upto 100 topics in each isolated network with di erent values of data locality to observe its e ect on the dissemination latency. For example, in case of 100 topics per isolated network and a data locality of .9, 90 topics are local to this isolated network and the remaining 10 topics are global (i.e have interested subscribers in some other networks). Figure 6 shows the end-to-end latency of data dissemination for increasing values of data locality for 60, 80 and 100 topics per isolated network. As seen in Figure 6, as the data locality increases, the end-to-end dissemination latency decreases since an increasingly large fraction of messages are local to an isolated network and do not incur WAN latencies for their dissemination thus validating the design of the Edge Broker Layer.

Scalability Results

We used a total of 400 VMs in our scalability experiments: 160 VMs were used for the brokers (120 VMs for edge brokers and 40 VMs for routing brokers) and of the remaining 240 VMs, 40 VMs were used for publishers and 200 VMs for subscribers. Each of these VMs runs either 25 publisher or 50 subscriber test applications. We placed 50 publishers or 100 subscribers in each network (i.e., 2 VMs for each network, thereby creating a total of 1000 publishers and 10,000 subscribers). Subscribers in each network are interested in 100 topics out of 1000 topics in the system.

Scalability of the Broker Overlay

Layer. Since the edge brokers are responsible for delivering data incoming from other brokers to subscribers in a local network, the computation overhead on edge brokers grows linearly as the number of adopted topics increases. Figure 7a and7d show the latency and Edge Broker CPU utilization results for increasing number of topics in each local network. e CPU utilization increases linearly with the number of adopted topics; the average and maximum latency values grow as well.

Our solution supports load balancing at the Routing Broker layer, allowing the system to scale with the number of topics in the system. Figure 7b and 7e present latency and CPU usage for di erent number of routing brokers. When the number of routing brokers is small, in this case 5, the CPU of the routing brokers becomes saturated and latency gets adversely impacted. However, a er autoscaling the number of routing brokers to 10, latency values improve.

e results in Figure 7e also validate that CPU usage decreases linearly by increasing the number of routing brokers.

Scalability of the Coordination Layer.

We evaluate the scalability of a ZooKeeper-based centralized coordination service by increasing the number of simultaneously joining subscribers from 2,000 to 10,000 in steps of 2,000. Figure 7c shows latency, i.e., the amount of time it takes for the server to respond to a client request, increases from 10ms to 20ms. We used mntr, a ZooKeeper monitoring service 7 , to retrieve the number of used znodes and watches. Figure 7f presents results for increasing number of used znodes and watches as the system scales. ese results show that the overhead of ZooKeeper based centralized coordination service remains acceptable even at scale. In our experiments, we have used zookeeper in standalone mode, however, as we increase the number of ZooKeeper servers in the ensemble, the latency overhead is expected to decrease further.

Deadline-aware Overlays

Deadline-aware overlays can be used for topics which have stricter data delivery requirements in case of congested, lossy and slow WAN links. To evaluate deadline-aware overlays in PubSubCoord, we compare the dissemination latency and broker overhead for deadline-aware multi-path vs single-path overlays. We used the topology shown in Figure 3 for our experiments. Verizon's8 delay and loss dataset was used for emulating WAN link behavior with Dummynet [START_REF] Rizzo | Dummynet: a simple approach to the evaluation of network protocols[END_REF]. We categorized delay and loss data into two groups (i.e., A with 30ms delay and no packet loss, and B with 250 msec delay and 1% packet loss in Table 1) and experimented with 8 possible combinations on the given links (i.e., L1, L2, and L3 as shown in Figure 3). ese test cases are described in Table 1.

Figure 8a, shows the average and maximum latencies for singlepath overlays for increasing values of emulated network latencies under both conditions-with and without packet loss. Figure 8b, shows the average and maximum latencies for deadline-aware multi-paths for all 8 cases as described in Table 1. We see that test cases 1 to 5 for multi-path overlays perform be er than any of cases for single-path overlays; and all cases of multi-path overlays outperform single-path overlay with 125 milliseconds network delay and 1% packet loss. Hence, topics with strict delivery requirements can bene t from deadline-aware overlays under adverse WAN link conditions. However, maintaining multi-path overlays does impose additional computation and network transfer overhead at the edge broker. We measured the network transfer overhead for 10,000 samples from a publisher to a subscriber for both single-path and multi-path overlays by using tcpdump 9 . Figure 8c, shows the additional network transfer overhead imposed by multi-path overlay.

RELATED WORK

Publish-Subscribe systems can be classi ed into three types (based on the trade-o between simplicity and performance on one hand and expressiveness on the other) [START_REF] Patrick | e Many Faces of Publish/Subscribe[END_REF] as follows: (1) Topic-based, (2) A ribute-based, and (3) Content-based systems. e most expressive form is Content-based publish-subscribe [START_REF] Barazzu | Elastic Scaling of a High-roughput Content-Based Publish/Subscribe Engine[END_REF], where subscriptions can be arbitrary boolean functions on the entire content of messages (e.g., XML documents). In A ribute-based pub-sub [START_REF] Li | A scalable and elastic publish/subscribe service[END_REF], messages are annotated with various a ributes and subscriptions are expressed as predicates over these a ributes. A publication matches a subscription if and only if all its a ribute values satisfy the corresponding predicates of the subscription. Topic-based pub-sub [START_REF] Gascon-Samson | Dynamoth: A Scalable Pub/Sub Middleware for Latency-Constrained Applications in the Cloud[END_REF]49] is the simplest form in which publishers tag their publications with a topic name; subscribers declare their interest by submi ing subscriptions to speci c topic names and all subscribers that have subscribed to a topic receive the message. Despite its simplicity, Topic-based pub-sub is widely used in industry, for example: Google Cloud Messaging (GCM) [START_REF]Google Cloud Messaging[END_REF], Amazon SNS [START_REF]Amazon Simple Noti cation Service[END_REF], Apache Ka a [START_REF] Kreps | Ka a: A distributed messaging system for log processing[END_REF], Apache Hedwig [2], Spotify [START_REF] Vinay Se Y | Hidden Pub/Sub of Spotify: (Industry Article)[END_REF], etc.

However, not many solutions support a variety of con gurable QoS policies [START_REF] Bellavista | ality of Service in Wide Scale Publish-Subscribe Systems[END_REF][START_REF] Carvalho | Scalable QoS-based event routing in publish-subscribe systems[END_REF] such as reliability, persistence, durability, deadline based delivery, security etc., which is much needed for IIoT application domains.

e OMG DDS [START_REF] Omg | Data Distribution Service speci cation[END_REF] is a Topic-based publish subscribe standard which supports low latency, peer-to-peer data delivery with a variety of con gurable QoS se ings. It has been deployed in many critical application domains such as healthcare 10 ,smart-grids 11 , etc. However, as noted earlier, current technology limitations restrict the use of DDS to a LAN.

To support WAN-scale data dissemination, pub/sub systems tend to form an overlay network of brokers. On the basis of the overlay topology, pub/sub systems can be categorized into: (1) Tree-based overlays [START_REF] Carzaniga | Design and Evaluation of a Wide-Area Event Noti cation Service[END_REF], (2) cluster-based overlays [START_REF] Cao | E cient event routing in content-based publish-subscribe service networks[END_REF], (3) structured/unstructured peer-to-peer overlays [START_REF] Aekaterinidis | PastryStrings: A Comprehensive Content-Based Publish/Subscribe DHT Network[END_REF][START_REF] Voulgaris | Sub-2-Sub: Self-Organizing Content-Based Publish Subscribe for Dynamic Large Scale Collaborative Networks[END_REF], and (4) cloud-based overlays [START_REF] Barazzu | Elastic Scaling of a High-roughput Content-Based Publish/Subscribe Engine[END_REF][START_REF] Gascon-Samson | Dynamoth: A Scalable Pub/Sub Middleware for Latency-Constrained Applications in the Cloud[END_REF][START_REF] Li | A scalable and elastic publish/subscribe service[END_REF][START_REF] Ma | A Scalable and Reliable Matching Service for Content-Based Publish/Subscribe Systems[END_REF][START_REF] Ma | Scalable and elastic event matching for a ribute-based publish/subscribe systems[END_REF][START_REF] Wang | A General Scalable and Elastic Content-Based Publish/Subscribe Service[END_REF]. In tree-based overlays, brokers are organized into a tree overlay. Tree-based overlays incur multi-hop routing latencies, lack recon guration exibility and impose costly maintenance of routing state information. In cluster-based overlays, brokers are organized into an unstructured overlay, where semantically similar brokers are grouped into the same cluster. However, due to frequent churn of nodes, maintaining semantic overlays leads to high latency. In peer-to-peer overlays, there are no dedicated brokers; instead each participating client is responsible for a small portion of the subscription space.

Although the peer-to-peer architecture is amenable to operate in wide-area networks with unreliable links and high node churn, it incurs high latency due to multi-hop routing and is not suitable for taking advantage of a well-engineered cloud environment where network links and server membership are much more stable. Cloud-based solutions o er low-latency, scalable, single-hop overlay routing of messages to all interested subscribers. However, most of the proposed solutions [START_REF] Li | A scalable and elastic publish/subscribe service[END_REF][START_REF] Ma | Scalable and elastic event matching for a ribute-based publish/subscribe systems[END_REF][START_REF] Wang | A General Scalable and Elastic Content-Based Publish/Subscribe Service[END_REF] are for a ribute-based pub/sub where they support only a xed a ribute space for participating clients. If there are multiple applications with di erent a ribute spaces, these cloud-based solutions do not o er a way in which multiple a ribute spaces can be supported simultaneously.

Dynamoth [START_REF] Gascon-Samson | Dynamoth: A Scalable Pub/Sub Middleware for Latency-Constrained Applications in the Cloud[END_REF] is similar to our solution but it provides a loadbalanced topic-based pub/sub service in the cloud with singlehop routing. Taking inspiration from recent edge/fog computing advances, PubSubCoord introduces a second Edge Broker layer, which allows us to gain latency bene ts for local data dissemination. Similarly, for edge/fog centric low-latency data dissemination, FogMQ [START_REF] Abdelwahab | FogMQ: A Message Broker System for Enabling Distributed[END_REF] supports online migration of message brokers to facilitate near-the-edge data analytics.

PubSubCoord adopts a two-layer architecture for low latency data dissemination; transparent and autonomous bridging of isolated DDS LANs and separation of local data dissemination concerns from the cloud. Like many cloud-based solutions, the Routing Broker layer in PubSubCoord allows us to use elastic cloud resources. Although architecturally di erent from pub/sub systems like Ka a [START_REF] Kreps | Ka a: A distributed messaging system for log processing[END_REF] and Hedwig [2], PubSubCoord also uses Zookeeper as its centralized coordination service.

Recent research including our prior work [START_REF] Hakiri | Supporting End-to-end Scalability and Real-time Event Dissemination in the OMG Data Distribution Service over Wide Area Networks[END_REF][START_REF] Jose | A content-aware bridging service for publish/subscribe environments[END_REF] has broadened the scope of DDS to WANs by bringing in routing engines to disseminate data from a local network to others. Many of these prior e orts require either invasive changes to existing applications or need speci c changes at di erent layers of the networking stack, which makes it hard to readily deploy such solutions. Our proposed solution utilizes similar routing engines introduced in the related work and additionally solves the automatic discovery and coordination problem between routing engines that otherwise requires signi cant manual e orts for large-scale systems.

CONCLUDING REMARKS

Emerging paradigms such as the Industrial Internet of ings illustrate the need to disseminate large volumes of data between a large number of heterogeneous entities that are geographically distributed, and require stringent QoS properties for data dissemination from the publishers of information to the subscribers. is paper presents the design, implementation, and evaluation of Pub-SubCoord, which is an autonomous and dynamic coordination and discovery service for WAN-scale pub/sub applications. PubSubCoord supports scalability in terms of data dissemination as well as coordination, autonomous discovery, and con gurable QoS properties.

e test harness and capabilities in PubSubCoord are available for download from www.dre.vanderbilt.edu/ ∼ kyoungho/pubsubcoord. A.1 OMG Data Distribution Service (DDS)

e OMG DDS speci cation de nes a distributed pub/sub communications standard [START_REF] Omg | Data Distribution Service speci cation[END_REF]. At the core of DDS is a data-centric architecture (i.e., subscriptions are de ned by topics, keyed data types, data contents, and QoS policies) for connecting anonymous data publishers with data subscribers in a logical global data space, as shown in Figure 9. e coupling between a publisher and subscriber is expressed in terms of topic name, its data type schema, and QoS a ributes of publishers and subscribers.

A domain is used to logically partition the global data space into groups that are isolated from each other within which the participants, i.e., publishers and subscribers can communicate. To ease the management, each publisher is made up of one or more DataWriters and each subscriber is made up of one or more DataReaders. Each DataWriter and DataReader can be associated with only one topic and perform the action of writing reading, respectively.

A Topic is a logical channel between DataWriters and DataReaders that speci es the data type of publication and subscription. e topic names, types, and QoS of DataWriters and DataReaders must match for them to communicate with each other.

A.2 OMG DDS QoS Policies

OMG DDS supports a number of di erent QoS policies that can be mixed and matched. Each QoS policy has o ered and requested semantics (i.e., o ered by publishers and requested by subscribers) and are used in conjunction with data types of topics to match pairs of endpoints, i.e., the DataReader and DataWriter. We brie y describe only those policies that we have used either in the design of PubSubCoord or in our empirical studies.

e reliability QoS controls the reliability of data ows between DataWriters and DataReaders at the transport level. It can be of two kinds:

and . e durability QoS speci es whether or not the DDS middleware stores and delivers previously published data samples to endpoints that join the network later. e reliability and persistency can be a ected by the history QoS policy, which speci es how much data must be stored in in-memory cache allocated by the middleware. Along with the history QoS policy, the lifespan QoS helps to control memory usage and lifecycle of data by se ing expiration time of the data on DataWriters, so that the middleware can delete expired data from the cache.

e deadline QoS policy speci es the deadline between two successive updates for each data sample. e middleware will notify the application via callbacks if a DataReader or a DataWriter breaks the deadline contract. Note that DDS makes no e ort to meet the deadline; it only noti es if the deadline is missed. e liveliness QoS speci es the mechanism that allows DataReaders to detect disconnected DataWriters. e ownership QoS speci es whether it allows multiple DataWriters to write data on a stream simultaneously. If it is set to have an exclusive owner, the exclusive owner is determined by the con gured strength values of DataWriters. e primary DataWriter with the highest strength is switched to a backup if it violates the deadline QoS or is disconnected.

A.3 DDS Routing Service

Since PubSubCoord relies on a broker-based architecture, we leverage and extend an existing DDS broker solution. Speci cally, we use the DDS Routing Service, which is a content-aware bridge service for connecting geographically dispersed DDS systems [START_REF] Jose | A content-aware bridging service for publish/subscribe environments[END_REF]. It integrates DDS applications across LANs as well as WANs. DDS Routing Service leverages all the entities of DDS so that it supports bene cial features already supported by DDS such as diverse QoS policies, content-based ltering, and dynamic type checking. In addition, it enables DDS applications to publish and subscribe data across domains in multiple networks without any changes to the applications.

A.4 ZooKeeper

ZooKeeper is a service for coordinating processes within distributed applications [START_REF] Hunt | ZooKeeper: wait-free coordination for internet-scale systems[END_REF]. e ZooKeeper service consists of an ensemble of servers that use replication to accomplish high availability with high performance and relaxed consistency. ZooKeeper provides the watch mechanism to notify a client of a change to a znode (i.e., a ZooKeeper data object containing its path and data content). ere exist many coordination recipes using ZooKeeper that are o en needed for distributed applications, such as leader election, group membership, and sharing con guration metadata. PubSubCoord exploits these capabilities in its design.

Figure 1 :

 1 Figure 1: PubSubCoord Architecture

Figure 2 :

 2 Figure 2: Specialized ZNode Tree for PubSubCoord system are rooted under the topics znode.e child znodes for every unique topic represent the endpoints, i.e., publishers and subscribers, associated with it. e leader znode is used to elect a leader among the routing brokers. A leader routing broker makes load balancing decisions for topic distribution among the cluster of routing brokers. e broker znode has child znodes for each routing broker where its location information (i.e., IP address and port number of a routing broker) is stored.e leader uses this information to associate a selected routing broker's location to a topic znode a er topic assignment.Dynamic changes in the system (e.g., broker or pub/sub endpoint join/leave, topic creation/deletion) are re ected in the creation, update, and deletion of znodes in the znode-tree. Brokers use the watch mechanism to set watches on interesting znodes to receive change noti cations and to take actions autonomously. PubSub-Coord exploits the ephemeral mode feature of ZooKeeper, where a speci c znode in the tree (and its subtree) is automatically deleted from the tree when the client session handling this znode is lost thereby providing automatic cleanup of state.

Figure 3 :

 3 Figure 3: Multi-path Deadline-aware Overlay Concept

Figure 4 :

 4 Figure 4: Routing Broker Sequence Diagram

Figure 6 :

 6 Figure 6: End-to-end Latency with increasing values of data locality

  End-to-end Latency of Pub/Sub by Di erent Number of Topics Per Network Per Edge Broker CPU Utilization (%) (d) CPU Utilization by Di erent Number of Topics Per Network

  (f) Number of ZNodes and Watches by Di erent Number of Joining Subscribers

Figure 7

 7 Figure 7: Scalability Experiments

Figure 8 :

 8 Figure 8: Deadline-aware Experiments

Figure 9 :

 9 Figure 9: DDS Architecture A DDS data publisher produces typed data streams identi ed by names called topics.e coupling between a publisher and subscriber is expressed in terms of topic name, its data type schema, and QoS a ributes of publishers and subscribers.A domain is used to logically partition the global data space into groups that are isolated from each other within which the participants, i.e., publishers and subscribers can communicate. To ease the management, each publisher is made up of one or more DataWriters and each subscriber is made up of one or more DataReaders. Each DataWriter and DataReader can be associated with only one topic and perform the action of writing reading, respectively.A Topic is a logical channel between DataWriters and DataReaders that speci es the data type of publication and subscription. e topic names, types, and QoS of DataWriters and DataReaders must match for them to communicate with each other.

  ep cache.get data() switch ep cache.get event type() do case child added if ! eb peer list.contains(ep eb locat or ) then eb peer list.add(ep eb locat or ) routin ser ice.add peer(ep eb locat or ) if ! topic list.contains(ep t opic ) then routin ser ice.create topic route(ep) topic multi set.add(ep t opic ) case child deleted topic multi set.delete(ep t opic ) if ! topic multi set.contains(ep t opic ) then eb peer list.delete(ep eb locat or ) routin ser ice.delete topic route(ep)

	Algorithm 1 Routing Broker Callback Functions
	function	(broker node cache)
	topic set = broker node cache.get data()
	for topic : topic set do
	if ! topic list.contains(topic) then
	ep cache = create children cache (topic)
	set listener(ep cache)
	topic list.add(topic)
	function	(ep cache)
	ep =

Table 1 :

 1 Deadline-aware Overlays Experiment Cases

	Test Cases	L1	L2	L3
	Case 1	A	A	A
	Case 2	A	A	B
	Case 3	A	B	A
	Case 4	A	B	B
	Case 5	B	A	A
	Case 6	B	A	B
	Case 7	B	B	A
	Case 8	B	B	B
	A = 30ms delay, no packet loss
	B = 250ms delay, 1% packet loss

8 (

 8 Aug 2015), 2100-2113. DOI:h p://dx.doi.org/10.1109/TPDS.2014.2346759 [49] Ye Zhao, Kyungbaek Kim, and Nalini Venkatasubramanian. 2013. DYNATOPS: A Dynamic Topic-based Publish/Subscribe Architecture. In Proceedings of the 7th ACM International Conference on Distributed Event-based Systems (DEBS '13). ACM, New York, NY, USA, 75-86. DOI:h p://dx.doi.org/10.1145/2488222. 2489273

	A BACKGROUND ON UNDERLYING
	TECHNOLOGIES USED IN PUBSUBCOORD
	Since we use the OMG DDS as the concrete pub/sub technology and
	ZooKeeper as the coordination service to describe PubSubCoord's
	contributions, this section provides an overview of these underlying
	technologies.

h ps://www.geso ware.com/sites/default/ les/Industrial Big Data Platform.pdf

Overload management and fault tolerance at the Edge Broker layer is an ongoing work.

h p://curator.apache.org znodes. We use the cache recipe to locally reserve data objects that are accessed multiple times for fast data access and reduce the load on ZooKeeper servers. e edge and routing brokers are entities we provided beyond existing DDS so ware entities. e edge brokers collocate themselves with RTI Connext's Routing Service, which provides edge-based routing capabilities. We have evaluated our design in an experimental testbed as we show later.

h ps://wiki.openstack.org/wiki/Neutron

h ps://community.rti.com/rti-doc/510/ndds.5.1.0/doc/pdf/RTI CoreLibrariesAndUtilities UsersManual.pdf

h ps://wiki.linuxfoundation.org/networking/netem

h p://zookeeper.apache.org/doc/trunk/zookeeperAdmin.html

h p://www.verizonenterprise.com/about/network/latency

h p://www.tcpdump.org

h ps://www.rti.com/industries/healthcare

h ps://www.rti.com/industries/energy

ACKNOWLEDGMENTS

is work is supported in part by NSF CAREER CNS 0845789. Any opinions, ndings, and conclusions or recommendations expressed in this material are those of the author(s) and do not necessarily re ect the views of NSF.


Tom Hohweiller

Nicolas Ducros

Françoise Peyrin

Bruno Sixou

Spectral CT material decomposition in the presence of poisson noise: A Kullback-Leibler approach

Keywords: Spectral CT, material decomposition, Kullback-Leibler distance, inverse problem

ou non, émanant des établissements d'enseignement et de recherche français ou étrangers, des laboratoires publics ou privés.

INTRODUCTION

Standard Computed Tomography (CT) is currently one of the most useful imaging modalities. However, X-ray detectors integrate the whole energy spectrum and since material attenuation depends on the incident X-ray photons energy, there is a loss of information. By performing energyresolved acquisition, spectral CT is promising to further improve the interest of CT. In particular, 3D mono-energetic images or 3D density map of materials can be retrieved. SPCT has many potentials for clinical applications, in particular, to image high Z contrast agents [START_REF] Feuerlein | Multienergy photoncounting k-edge imaging: potential for improved luminal depiction in vascular imaging 1[END_REF][START_REF] Cormode | Atherosclerotic plaque composition: analysis with multicolor ct and targeted gold nanoparticles 1[END_REF]. While, dual-energy CT have been proposed to acquire energy-resolved data [START_REF] Johnson | Dual-energy ct: general principles[END_REF], but with a large irradiation dose. Photoncounting detectors based on a new technology can count photons that are hitting the detector and retrieve their energy [START_REF] Taguchi | Vision 20/20: Single photon counting x-ray detectors in medical imaging[END_REF].

Three different approaches for decomposing SPCT data into material volumes are available. A first approach consists in the tomographic reconstruction of the energy-resolved measurements followed by a material decomposition step in the object domain [START_REF] Maaß | Image-based dual energy ct using optimized precorrection functions: A practical new approach of material decomposition Figure 2: Ground truth of the material map of soft tissues and bones (left), decomposition with WLS (middle) and with KL (right) for N 0 = 10 2.2 photons for α in image domain[END_REF]. A second approach is to do the tomographic reconstruction and the decomposition at the same time, which is usually referred to as the one-step approach [START_REF] Long | Multi-material decomposition using statistical image reconstruction for spectral ct[END_REF]. A third approach is to perform material decomposition in the projection domain followed by a tomographic reconstruction step [START_REF] Schlomka | Experimental feasibility of multienergy photon-counting k-edge imaging in pre-clinical computed tomography[END_REF]. The last approach is highly parallelizable (each projection angle can be processed individually) and it embeds all the non-linearities of the forward problem, i.e., tomographic reconstructions can then be performed linearly without any simplifying assumptions. In this work, material decomposition in the projection domain is considered.

Material decomposition is a non-linear ill-posed inverse problem that can be solved by minimizing a cost function composed of a data fidelity term and a regularization term. A regularization parameter is chosen to balance the two terms of the cost function. SPCT data are corrupted by photonic noise, which is Poisson distributed. Roessl et al. and Schlomka et al. have proposed decomposition methods that take into account the statistics of the noise in the projection domain [START_REF] Roessl | Cramér-rao lower bound of basis image noise in multiple-energy x-ray imaging[END_REF][START_REF] Schlomka | Experimental feasibility of multienergy photon-counting k-edge imaging in pre-clinical computed tomography[END_REF], e.g., using negative log likelihood. However, the latter methods rely on zero-th or first-order minimization algorithms [START_REF] Schirra | Spectral ct: a technology primer for contrast agent development[END_REF][START_REF] Xu | Sparsity-regularized image reconstruction of decomposed k-edge data in spectral ct[END_REF] that have slow convergence rates. Recently, a fast method based on a Gauss-Newton algorithm was proposed [START_REF] Ducros | Regularization of nonlinear decomposition of spectral x-ray projection images[END_REF]. Unfortunately, it uses a weighted least squares fidelity term that is not perfectly adapted to Poisson noise. In this work, the Kullback-Leibler distance is proposed to better handle the noise statistics and a Gauss-Newton algorithm is implemented to get fast decomposition.

This paper is organized as follows. In Section II, the forward model that maps the projected mass of the material onto the number of photons detected is presented. In Section III, we detail the regularization functional considered and the minimization algorithms used. Numerical experiments are presented in Section IV before the conclusion.

THEORY

Object model

Assuming that the object of interest can be decomposed on a basis of M materials [START_REF] Alvarez | Energy-selective reconstructions in x-ray computerised tomography[END_REF], the linear attenuation coefficient is written

µ(E, x) = M m=1 ρ m (x)τ m (E) (1)
where µ(E, x) is the linear attenuation coefficient (in cm -1 ) at point x and energy E, ρ m (x) is the mass density (in g.cm -3 ) of the m-th material at point x, and τ m (E) is the mass attenuation (in cm 2 .g -1 ) of the m-th material.

Forward model

The standard SPCT forward model [START_REF] Ducros | Regularization of nonlinear decomposition of spectral x-ray projection images[END_REF][START_REF] Roessl | Sensitivity of photoncounting based-edge imaging in x-ray computed tomography[END_REF] is written as

s i (u) = R d i (E)n 0 (E, u) exp - M m=1 a m (u)τ m (E) dE (2) where a m (u) = L(u) ρ m (x)dx (3)
is the projection of all the mass of the m-th material along the x-ray path L(u), s i (u) is the number of photons hitting the detector in the i-th energy bin at pixel u, n 0 (E, u) is the emitted X-ray spectrum, and d i (E) is the response function of the i-th energy bin of the detector [START_REF] Schlomka | Experimental feasibility of multienergy photon-counting k-edge imaging in pre-clinical computed tomography[END_REF].

Discretization and noise

The detector is assumed to have P pixels, with u p the location of the p-th pixel, and I energy bins. We define the measurement vector s ∈ R IP as

s = [s 1,1 , . . . s i,p , . . . s I,P ] (4)
where s i,p is the photon count measured in the i-th energy bin at pixel u p , i.e., s i,p = s i (u p ). Similarly, the projected mass density (PMD) vector a ∈ R M P is defined by

a = [a 1,1 , . . . a m,p , . . . a M,P ] (5)
where a m,p is PMD of the m-th material at pixel u p , i.e., a m,p = a m (u p ). Data is assumed to be corrupted by Poisson noise. Let s * be the noiseless data vector, s the noisy data, and P(γ) the Poisson distribution with mean γ. We have:

s ∼ P(γ = s * ). ( 6
)

Inverse problem

Material decomposition is an ill-posed problem that consists in recovering the projected mass density a from the data s. Let s = F(a) be the forward model defined in [START_REF] Cormode | Atherosclerotic plaque composition: analysis with multicolor ct and targeted gold nanoparticles 1[END_REF]. The PMD is recovered by minimizing the following cost function

C(s, F(a)) = D(s, F(a)) + αR(a) ( 7
)
where D(s, F(a)) is the fidelity term, α the regularization parameter and R(a) the regularization term used to stabilize the solution. The Tikhonov regularization terms used in [START_REF] Ducros | Regularization of nonlinear decomposition of spectral x-ray projection images[END_REF] is chosen

R(a) = ||∆a soft || 2 2 + ||∇a bone || 2 2 ( 8
)
where ∆ and ∇ are first-and second-order differential operator. 2

PROPOSED MATERIAL DECOMPOSI-TION

Fidelity term

The usual fidelity term is the WLS, defined by

D WLS (s, F(a)) = ||W(s -F(a))|| 2 (9)
where ||.|| 2 is the 2 -norm and W is the weighting matrix defined as W = diag( 1 √ s+1 ) where diag(x) represents a diagonal matrix with the elements x. However the noise associated with photon detection problem is Poisson noise. The Kullback-Leibler divergence is the distance associated to Poisson noise [START_REF] Resmerita | Joint additive kullback-leibler residual minimization and regularization for linear inverse problems[END_REF]. The KL distance between two Poissondistributed data s and F(a) is written [START_REF] Kullback | On information and sufficiency[END_REF]

D KL (s, F(a)) = i (s i +ζ) log s i + ζ F(a) i + ζ +F(a) i -s i ( 10
)
where ζ is a small integer that avoids to dividing by zero and having the log of zero.

The corresponding data fidelity terms will be denoted D KL and D WLS .

Minimization of the cost function

The cost function ( 7) is minimized by using a Gauss-Newton's algorithm. This second-order algorithm provides faster convergence than first order methods. Starting with an initial guess a 0 , the solution is computed iteratively with the following update formula

a k+1 = a k + λ k δa k (11)
where a k is the solution at the iteration k, δa k is the Newton's step and λ k a scalar optimizing the minimization in the direction δa k . The Newton's step is computed by solving

(J Z h J+αH reg )δa k = -(J Z g (F(a k )-s)+αg reg ) ( 12
)
where J is the Jacobian of the forward problem, H reg and g reg is the Hessian and the gradient of the regularization term, respectively. Z h and Z g are diagonals matrices which depend on the fidelity term. They can be written as follow [START_REF] Favati | Perfor-mance analysis of maximum likelihood methods for regularization problems with nonnegativity constraints[END_REF] WLS :

Z g = 2W T W Z h = 2W T W , KL : Z g = Y -1 Z h = Y -2 S
where W is the weighting matrix of the WLS term, Y = diag(F(a) + ζ) and S = diag(s + ζ).

The step length λ k is computed as

λ k = argmin λ C(s, F(a k + λδa k )). ( 13
)

NUMERICAL EXPERIMENTS AND IMAGE ANALYSIS

Material decomposition is performed in the numerical phantom DigiMouse [START_REF] Dogdas | Digimouse: a 3d whole body mouse atlas from CT and cryosection data[END_REF]. Projections are generated with the radon Matlab function. The goal is to decompose M = 2 materials (soft tissues and bones) from projections of P = 437×992 pixels.

We define the number of photons N 0 send at each pixel detector

N 0 = R n 0 (E, u)dE ( 14
)
Since N 0 controls the signal-to-noise ratio, a wide range is tested: N 0 = [10 2 , 10 2.2 , ..., 10 4 ]. The Xray spectrum used is generated with a 90 kV source. Considering a detector with I = 3 energy bins (14-39keV , 40 -64keV and 65 -90keV as energy bins).

The regularization parameter values tested are α = [10 -2 , 10 -1.8 , ..., 10 4 ]. Our initial guess a 0 was set as uniform material maps with a 0 soft = 2 g.cm -2 and a 0 bone = 0 g.cm -2 . The algorithm is stopped when the relative decrease of the cost function is below 0.1%, if the computed step λ k is below 10 -2 or if the maximum iteration number 50 is reached. The relative error is computed as

ξ = M m=1 ||a dec m -a true m || 2 ||a true m || 2 (15)
where a dec m is the decomposed map of the m th material and a true m is the ground truth for the material m. Moreover, α denotes the regularization parameter that minimizes the relative error ξ

α = argmin α ξ(α) (16)

RESULTS AND DISCUSSION

In this section, we present results from experiments on DigiMouse.

In Figure 1 (left), the evolution of the fidelity terms D KL and D WLS is plotted as a function of the number of iterations. There is an apparent decrease of the regularized functionals, and local minima are obtained after four iterations. The algorithm stops since there is a low decrease of the cost function or a small step length λ k , which represents a local minimum because δa k is, by construction, a descent's direction. Figure 1 (right) shows the evolution of the decomposition error for α with increasing numbers of photons N 0 for the fidelity terms D KL and D WLS . For real projections, the ground truth is not known and thus from this figure the improvement of the reconstruction results with the KL distance for the lower number of photons is highlighted. For these low numbers of photons the weighted least square is not a good approximation of the KL distance and this statistical divergence is more efficient to take into account the noise statistic. Similar curves are obtained when relative errors for a given material are considered.

Figure 2 shows the ground truth images of the PMD images and decompositions with WLS and KL for N 0 = 10 2.2 . A low number of photons is chosen to highlight the performance of KL. The KL distance clearly retrieves more faithfully soft tissues and bones than WLS at a small number of photons.

In Table 1, the average time required in order to make one iteration (3.5 GHz Xeon E5-1650 v3 and 64 GiB of RAM) as well as the standard deviation for α = [10 -2 , 10 -1.8 , ..., 10 4 ] are shown for N 0 = 10 2.2 . The minimum and the maximum number of iterations required for convergence are also displayed. Table 2 provides the value of the optimal α, the computation time and the iterations number in order to converge for N 0 = 10

CONCLUSION

In this work, methods for material decomposition are carried out in the projection domain in SPCT. The non-linearity of the forward problem is taken into account. Two data fidelity terms, one based on weighted least squares and the other on the Kullback-Leibler divergence, are compared. The regularized cost functions are minimized with a Gauss-Newton scheme. The methods are compared for 2D projections of a 3D realistic phantom for different numbers of photons. The Kullback-Leibler distance gives the best results for the lower number of photons. In the future, we plan to assess the method with other phantoms and a larger number of materials and to develop methods for automatic selection of the regularization parameter.

Figure 1 :

 1 Figure 1: Evolution of the fidelity terms for α and for N 0 = 10 2.2 (left) and of the decomposition error with the number of photons (right)

Table 1 :

 1 Mean and standard deviation of time and the minimum and maximum of iterations to converge for n 0 = 10 2.2 with α = [10 -2 , 10 -1.8 , ...,10 4 ]

Table 2 :

 2 2.2 . Both KL and WLS converge rapidly for a large range of α's and take almost the same CPU time. KL is found to outperform WLS for a low numbers of photons. Value, time and number of iteration require to converge for α at n 0 = 10 2.2

		α	Time (sec) Iteration number
	WLS	10 2	2.74	4
	KL	10 1.4	2.47	4

ACKNOWLEDGEMENT

This work was supported by the LABEX PRIMES (ANR-11-LABX-0063) of Université de Lyon, within the program "Investissements d'Avenir" (ANR-11-IDEX-0007) operated by the French National Research Agency (ANR). It was also supported by the ANR grant ANR-11-INBS-0006.

Conflict of interest: none

BIBLIOGRAPHY