Patent Description:
Messages to all endpoints are published at the same rate, so the slowest endpoint determines the overall rate of message publishing and the complete pipeline halts if any one endpoint is unresponsive. The latter scenario may be addressed by ignoring endpoints which have been unresponsive for a specified period of time (e.g., one hour), but such an approach does not address the problems caused by slow-but-functioning endpoints.

One option to improve overall latency is to create a separate pipeline per endpoint. This option would prevent one endpoint from hindering the transmission of messages to another endpoint. However, since each pipeline reads and pre-processes all received messages, each additional pipeline requires duplicative reading and pre-processing resources. Each additional pipeline also requires additional bookkeeping overhead to avoid sending duplicate messages.

Systems are needed to provide improved publish-to-delivery latency for messages delivered to one or more of multiple endpoints. Such systems may reduce the effect of a slow or failed endpoint on the delivery of messages to other endpoints without consuming an unsuitable amount of available computing resources.

<CIT> describes techniques for using a message spinning engine to create and/or manage messaging queues in a distributed network using virtual servers. An abstraction layer formed by virtualized servers may enable the message spinning engine to seamlessly transition messaging queues while minimizing the likelihood of exceeding the parameters of existing service level agreements. The message spinning engine may include a service mapping module to maintain mappings between source business applications and virtualized servers, a product bridge to implement message encapsulation for communication between different messaging queue products, and a messaging queue monitoring console to analyze performance and capacity of physical services and messaging services and accordingly adjust levels of service for source business applications.

The invention relates to a system and a method as set forth in the appended claims. It will be understood that aspects of this disclosure falling outside the scope of the claims may not be part of the invention but may be useful to understand the invention.

The following description is provided to enable any person in the art to make and use the described examples. Various modifications, however, will be apparent to those in the art.

Embodiments address the foregoing by dynamically adding and removing routing pipelines based on endpoint performance. Embodiments create a separate routing pipeline for each of one or more classes of low-performing endpoints and assign endpoints to the various routing pipelines based on monitored endpoint performance. An endpoint may be re-assigned from one pipeline to another if the performance of the endpoint changes. Embodiments therefore contain the effect of slow endpoints to one or more routing pipelines while well-performing endpoints are served by a dedicated routing pipeline. A limited and configurable number of overall routing pipelines ensures efficient resource consumption.

Examples may employ any suitable system to measure endpoint performance. In some examples, the generation of pipelines and assignment of endpoints to pipelines are based on predictions of future endpoint performance, which may in turn be based on current and historical endpoint performance. In one non-exhaustive example, some examples measure endpoint performance by tracking an exponential moving average of the time taken to write a message to the endpoint. If writing a message to an endpoint is asynchronous (e.g., through a buffer or any other async mechanism), endpoint performance may be measured by tracking an exponential moving average the latency of initiating the call (i.e., buffer wait time).

Some examples further provide a catch-up routing pipeline. The catch-up routing pipeline allows an endpoint to catch-up to the message checkpoint position of its new higher-performance pipeline before joining the higher-performance pipeline. The catch-up pipeline also enables the system to confirm the improved performance of the endpoint before assigning the endpoint to the higher-performance pipeline. Operation of the catch-up routing pipeline will be described below.

According to some examples, routing service customers may be offered endpoint prioritization (i.e., fixed assignment to a default routing pipeline) and/or endpoint grouping (i.e., fixed assignment of one or more endpoints to a particular pipeline), perhaps for appropriate additional fees. A customer may be provided with the ability to manually upgrade to downgrade endpoints to particular routing pipelines based on contextual information such as an endpoint maintenance window and known downstream problems.

<FIG> is a block diagram of system <NUM> to describe an example of a generic example. System <NUM> includes routing component <NUM>, incoming messages <NUM>, <NUM> and <NUM>, and endpoints <NUM>, <NUM> and <NUM>. Generally, routing component <NUM> receives messages <NUM>, <NUM> and <NUM> and routes each message to an appropriate one of endpoints <NUM>, <NUM> and <NUM>.

Messages <NUM>, <NUM> and <NUM> may comprise any type of messages conforming to any type of protocol and containing any type of payload that is or becomes known. Messages <NUM>, <NUM> and <NUM> may also be received from any one or more message sources that are or become known. Similarly, endpoints <NUM>, <NUM> and <NUM> may comprise any type of computing message endpoints that are or become known.

Routing component <NUM> includes routing pipelines <NUM> and <NUM>, although examples are not limited to two routing pipelines. Each of routing pipelines <NUM> and <NUM> is assigned to deliver messages to one or more of endpoints <NUM>, <NUM> and <NUM>. The assignment is based on the relative performance of each endpoint. As shown, routing pipeline <NUM> delivers messages to endpoints <NUM> and <NUM>, whose performance is determined to be above a threshold, and pipeline <NUM> delivers messages to endpoint <NUM>, whose performance is determined to be below the threshold.

In operation, each of routing pipelines <NUM> and <NUM> receives each of messages <NUM>, <NUM> and <NUM>, reads and pre-processes each of messages <NUM>, <NUM> and <NUM>, and delivers appropriate ones of messages <NUM>, <NUM> and <NUM> to its assigned endpoint(s). For example, it will be assumed that message <NUM> is to be delivered to all endpoints, message <NUM> is to be delivered to endpoint <NUM>, and message <NUM> is to be delivered to all endpoint <NUM>. Accordingly, routing pipeline <NUM> delivers message <NUM> to endpoints <NUM> and <NUM> and delivers message <NUM> to endpoint <NUM>, and routing pipeline <NUM> delivers message <NUM> and message <NUM> to endpoint <NUM>.

Routing component <NUM>, endpoints <NUM>, <NUM> and <NUM>, and each other component described herein may be implemented by one or more computing devices (e.g., computer servers), storage devices (e.g., hard or solid-state disk drives), and other hardware as is known in the art. The components may be located remote from one another and may be elements of one or more cloud computing platforms, including but not limited to a Software-as-a-Service, a Platform-as-a-Service, and an Infrastructure-as-a-Service platform. According to some examples, each routing pipeline <NUM> and <NUM> is implemented by a dedicated virtual machine.

<FIG> is a block diagram of a more detailed example according to some examples. System <NUM> includes hub <NUM>, IoT devices <NUM>-<NUM>, and endpoints <NUM>, <NUM> and <NUM>. Hub <NUM> receives messages from IoT devices <NUM>-<NUM> and routes appropriate ones of messages to one or more of endpoints <NUM>, <NUM> and <NUM>. Implementations may include a large number (e.g., hundreds, thousands or more) of IoT devices. Endpoints <NUM>, <NUM> and <NUM> may comprise an external event hub, external storage, a distributed database, etc..

Hub <NUM> includes gateway <NUM> to receive messages from IoT devices <NUM>-<NUM> and routing service <NUM> to route the received messages. Routing service <NUM> uses routing policy to determine the endpoint(s) to which a particular message should be delivered. According to routing policy <NUM>, the determination may be based on message source, message type, message content and/or any other factors that are or become known. Some messages received from IoT devices <NUM>-<NUM> may be consumed at an event hub of hub <NUM> and not delivered to external endpoints.

Each of routing pipelines <NUM> and <NUM> applies routing policy <NUM> to each message received at gateway <NUM> and, if routing policy <NUM> indicates that a received message should be delivered to one or more endpoints assigned to the pipeline, the pipeline delivers the message to the one or more endpoints. Routing pipelines <NUM> and <NUM> may therefore ignore routing policies which are not relevant to any of their assigned endpoints.

According to the <FIG> example, endpoint <NUM> has been identified as being slow to process incoming messages. In some examples, endpoint <NUM> is not currently processing messages slowly but is predicted to be slow in the near future. Techniques for measuring and predicting endpoint performance will be described below. Endpoint <NUM> is therefore assigned to "slow" routing pipeline <NUM>, while better-performing endpoints <NUM> and <NUM> are assigned to "default" routing pipeline <NUM>. Accordingly, routing pipeline <NUM> delivers messages to endpoints <NUM> and <NUM>, and pipeline <NUM> delivers messages to endpoint <NUM>. Consequently, the slow performance of endpoint <NUM> does not negatively affect the delivery of messages to endpoints <NUM> and <NUM>.

Generation of additional performance-related routing pipelines according to some examples will now be described. System <NUM> is again illustrated in <FIG>, prior to the instantiation of pipeline <NUM>. <FIG> also illustrates checkpoint store <NUM>, in communication with routing service <NUM> and each of endpoints <NUM>, <NUM> and <NUM>.

Each routing pipeline maintains a checkpoint of its last successfully-delivered message. In case of a power failure, the routing pipeline only has to resend those messages which were processed after the checkpoint. A routing pipeline moves its checkpoint forward only after a message has been processed by each endpoint for which it is intended.

Each endpoint also keeps a checkpoint of the last message it successfully processed. In this regard, each endpoint may include a message buffer to store received but not-yet-processed messages.

<FIG> is a flow diagram of process <NUM> to re-assign endpoints to routing pipelines based on endpoint performance according to some examples. Process <NUM> and the other processes described herein may be performed using any suitable combination of hardware or software. Executable code embodying these processes may be executed by a central processing unit of a microprocessor or microcontroller, for example, and may be stored in any non-transitory tangible medium, including a read-only memory, a volatile or non-volatile random access memory, a fixed disk, a DVD, a Flash drive, or a magnetic tape. Examples are not limited to the examples described below.

It will be assumed that system <NUM> is operating prior to S410 to receive messages from IoT devices <NUM>-<NUM> and deliver them to appropriate ones of endpoints <NUM>, <NUM> and <NUM>. As described above, such delivery results in updating of checkpoints respectively associated with routing pipeline <NUM>, endpoint <NUM>, endpoint <NUM> and endpoint <NUM>.

Process <NUM> pauses at S410 for a preconfigured interval before continuing to S420. As will be appreciated from the foregoing description, the interval is specified to prevent continuous evaluation and re-assigning of endpoints among routing pipelines. According to some examples, process <NUM> starts with an initial delay to capture a first x minutes of endpoint performance information and then repeats every <NUM> seconds.

At S420, each existing endpoint is evaluated to determine whether the endpoint is unhealthy or slow. Endpoint performance may be determined based on many different metrics such as write/publish latency, failure rate, etc. As described above, each endpoint may be associated with a small in-memory buffer to absorb endpoint performance fluctuations. Some examples may monitor, for each endpoint, a duration during which the buffer is full. This duration (i.e., "buffer wait time") corresponds to a duration over which a routing pipeline assigned to the endpoint is blocked because no more messages can be enqueued in the endpoint's buffer.

The determination at S420 may be performed using a prediction model which is based on the buffer wait time. For example, an exponential moving average may be determined for each endpoint at S420. One example model is EMAnext = α · EMAt + (<NUM>- α) · EMAt-<NUM>. Such a model may be useful because only one buffer wait time value needs to be stored for each endpoint, as opposed to many past values which would be required to calculate a true moving average. A prediction model according to some examples may comprise a Markov decision process, or a machine learning neural network which is trained based on historical endpoint performance data and inputs which may include, but are not limited to, time, customer configuration, region of customer, and endpoint type.

The determination at S420 may consist of determining whether the performance metric (e.g., the predicted buffer wait time) reflects significantly worse performance than other endpoints assigned to the same routing pipeline. For example, in the case of <FIG>, it is determined whether the performance of any one of endpoints <NUM>, <NUM> and <NUM> is significantly worse than the performance of the other two endpoints. In this regard, endpoint slowness is measured relative to the slowness of all other endpoints sharing a same routing pipeline.

According to some examples, an endpoint is considered slow if its performance is more than 10x worse than the best-performing endpoint of its same routing pipeline. In some cases, pipeline assignment may be based on a linear scale (e.g., endpoints exhibiting <NUM> to n times worse latency of best-performing endpoint assigned to a default pipeline, n to In assigned to a slow pipeline, all others assigned to a stuck pipeline). In other cases, pipeline assignment may be based on an exponential scale (e.g., endpoints exhibiting <NUM> to n times worse latency of best-performing endpoint assigned to a default pipeline, n to n<NUM> assigned to a slow pipeline, all others assigned to a stuck pipeline). Flow returns to S410 to wait if none of endpoints <NUM>, <NUM> and <NUM> are determined to be slow at S420.

According to some examples, each endpoint is also associated with an Unhealthy Since timestamp (e.g., stored in checkpoint store <NUM>). The timestamp is updated each time a message is successfully received. Accordingly, if an endpoint has been unable to accept messages for a long time, a large difference will exist between its Unhealthy Since timestamp and the current time. In such examples, S420 may also include a determination of whether an endpoint is unhealthy based on its Unhealthy Since timestamp.

It will now be assumed that endpoint <NUM> is determined to be slow at S420. Flow therefore proceeds to S430 to determine whether routing service <NUM> includes a routing pipeline associated with lower-performance endpoints. With respect to the <FIG> example, the determination at S430 is negative and flow therefore proceeds to S440.

A new routing pipeline is instantiated at S440. The new routing pipeline is intended to deliver incoming messages to a lower-performing group of one or more endpoints. According to some examples, S440 includes halting and restarting routing service <NUM> with a new configuration including an additional pipeline such as pipeline <NUM> of <FIG>. As shown, new routing pipeline <NUM> is assigned to endpoint <NUM> at S450 and routing pipeline <NUM> remains assigned to endpoints <NUM> and <NUM>. The checkpoint values of pipeline <NUM> and endpoints <NUM> and <NUM> remain as they were prior to the restart. New pipeline <NUM> is assigned the checkpoint of the routing pipeline (i.e., routing pipeline <NUM>) from which endpoint <NUM> was re-assigned.

Flow returns to S410 to again pause for a preconfigure audit interval and continue as described above. S420 is performed for each endpoint of each routing pipeline and as described above, the performance of an endpoint is evaluated only against the performance of other endpoints in its pipeline. Accordingly, if two endpoints are assigned to routing pipeline <NUM> and the performance of one of the endpoints is significantly worse than the other, a third pipeline may be instantiated at S440 and the worse-performing endpoint may be assigned thereto at S450. Additional pipelines may be added for increasingly poor-performing endpoints. In another example, if it is determined that the performance of endpoint <NUM> is significantly worse that the performance of endpoint <NUM>, endpoint <NUM> may be re-assigned to pipeline <NUM> at S450.

According to some examples, instantiation of a second routing pipeline at S440 includes instantiation of a third "catch-up" pipeline such as pipeline <NUM> of <FIG>. In operation, an endpoint is re-assigned to catch-up pipeline <NUM> if a sufficient improvement in its performance is detected. Only one endpoint is assigned to the catch-up pipeline at any given time. Catch-up pipeline <NUM> is used to confirm that an endpoint has actually recovered so it isn't quickly re-assigned back to a lower-performance pipeline after being assigned to the default pipeline. Catch-up pipeline <NUM> also allows the recovered endpoint to catch up with the message queue before being re-assigned to the default pipeline.

In some examples, the periodic determination at S420 may include evaluation of whether the performance of the endpoint associated with the catch-up routing pipeline has improved to a sufficient degree. The determination may be based on any of the above-mentioned performance measures. Moreover, the performance of the endpoint may be compared to the performance of the endpoints of the default routing pipeline to determine whether the performance has improved to a sufficient degree (e.g., better than 10x worse than the best-performing endpoint of the default routing pipeline). <FIG> illustrates reassignment of endpoint <NUM> from "slow" routing pipeline <NUM> to catch-up pipeline <NUM> after a determination that the performance of endpoint <NUM> has sufficiently improved. Reassignment includes updating the checkpoint of routing pipeline <NUM> to equal the current checkpoint of endpoint <NUM>.

<FIG> is a flow diagram of a process to re-assign an endpoint from a catch-up routing pipeline to a default routing pipeline according to some examples. Process <NUM> may be performed with respect to a catch-up routing pipeline and not for any of the other pipelines of routing service <NUM>. After waiting for an audit interval at S810 (which may be equal to or different from the other audit intervals discussed herein), it is determined at S820 whether the improved performance of the endpoint of the catch-up pipeline has been sustained for a suitable amount of time. This amount of time may be preconfigured and may be selectable by the customer. If not, flow returns to S810.

If the determination at S820 is positive, it is determined whether the checkpoint of the catch-up pipeline is close to the checkpoint of the default pipeline. Any suitable measure of "closeness" may be used, for example, <NUM> messages. Flow pauses at S830 until the determination is positive, at which point the endpoint is re-assigned from the catch-up pipeline to the default pipeline.

Many variations of the features described herein are possible. For example, any number of pipelines may be employed in addition to a default pipeline and, in some example, a catch-up pipeline, with each pipeline associated with a respective performance grouping. <FIG> illustrates system <NUM> including four routing pipelines. The example shows default pipeline <NUM>, slow pipeline <NUM>, catch-up pipeline <NUM> and "stuck" pipeline <NUM>. Stuck pipeline <NUM> is assigned to endpoint <NUM>.

In some examples, stuck pipeline <NUM> is assigned to endpoints which have been unhealthy (e.g., as indicated by their Unhealthy Since parameter) for at least a given amount of time (e.g., one hour). An endpoint may move from stuck pipeline <NUM> to catch-up pipeline <NUM> if a performance improvement is detected, just as an endpoint may be moved from slow pipeline <NUM> to catch-up pipeline <NUM>. In some examples, pipeline <NUM> is a "slower" pipeline which is assigned to endpoints whose performance is at least 10x worse than the performance of the best-performing endpoint assigned to slow endpoint <NUM>.

<FIG> is a block diagram of system <NUM> according to some examples. System <NUM> may comprise a general-purpose computing server and may execute program code to provide a routing service as described herein. System <NUM> may be implemented by a cloud-based virtual server according to some examples.

System <NUM> includes processing unit <NUM> operatively coupled to communication device <NUM>, persistent data storage system <NUM>, one or more input devices <NUM>, one or more output devices <NUM> and volatile memory <NUM>. Processing unit <NUM> may comprise one or more processors, processing cores, etc. for executing program code. Communication interface <NUM> may facilitate communication with external devices, such as client devices, and data providers as described herein. Input device(s) <NUM> may comprise, for example, a keyboard, a keypad, a mouse or other pointing device, a microphone, a touch screen, and/or an eye-tracking device. Output device(s) <NUM> may comprise, for example, a display (e.g., a display screen), a speaker, and/or a printer.

Data storage system <NUM> may comprise any number of appropriate persistent storage devices, including combinations of magnetic storage devices (e.g., magnetic tape, hard disk drives and flash memory), optical storage devices, Read Only Memory (ROM) devices, etc. Memory <NUM> may comprise Random Access Memory (RAM), Storage Class Memory (SCM) or any other fast-access memory.

Routing service <NUM> may comprise program code executed by processing unit <NUM> to cause system <NUM> to instantiate and manage routing pipelines which receive, pre-process, and deliver messages as described herein. Routing policies <NUM> may determine how and where messages are routed, checkpoints <NUM> may include current checkpoints of routing pipelines and endpoints, and performance information <NUM> may include data indicating the current and/or predicted performance of the endpoints as described herein. Data storage device <NUM> may also store data and other program code for providing additional functionality and/or which are necessary for operation of system <NUM>, such as device drivers, operating system files, etc..

The foregoing diagrams represent logical architectures for describing processes according to some examples, and actual implementations may include more or different components arranged in other manners. Other topologies may be used in conjunction with other examples. Moreover, each component or device described herein may be implemented by any number of devices in communication via any number of other public and/or private networks. Two or more of such computing devices may be located remote from one another and may communicate with one another via any known manner of network(s) and/or a dedicated connection. Each component or device may comprise any number of hardware and/or software elements suitable to provide the functions described herein as well as any other functions.

Claim 1:
A computer-implemented system (<NUM>; <NUM>) comprising:
a routing service configured to:
monitor or predict a performance of a plurality of endpoints; and
assign a first plurality of the plurality of endpoints to a first routing pipeline and a second plurality of the plurality of endpoints to a second routing pipeline based on the monitored or predicted performance of the plurality of endpoints, wherein the second plurality of endpoints are associated with lower performance than the first plurality of endpoints; and wherein
the routing service comprises:
the first routing pipeline (<NUM>; <NUM>) configured to receive messages and to route a first plurality of the messages to the first plurality of endpoints, wherein the first routing pipeline (<NUM>; <NUM>) is configured to move a message checkpoint forward after each message of the first plurality of the messages has been processed, wherein the processing of each message of the first plurality of messages is by each of the first plurality of endpoints for which it is intended; and
the second routing pipeline (<NUM>; <NUM>) configured to receive the messages and to route a second plurality of the messages to the second plurality of endpoints, wherein the second routing pipeline (<NUM>; <NUM>) is configured to move a message checkpoint forward after each message of the second plurality of the messages has been processed, wherein the processing of each message of the second plurality of messages is by each of the second plurality of endpoints for which it is intended.