Efficient access scheduling for super scaled stream processing systems

• US 10,409,650 B2
• Filed: 02/24/2016
• Issued: 09/10/2019
• Est. Priority Date: 02/24/2016
• Status: Active Grant

• Current Assignee: Salesforce.com, Inc.
• Original Assignee: Salesforce.com, Inc.
• Inventors: Zhao, Yingwu, Hildebrand, Hal Scott
• Primary Examiner(s): Hasan, Syed H
• Application Number:
  US15/052,822
• Publication Number:
  US 20170242887A1
• Time in Patent Office: 1,294 Days
• CPC Class Codes: G06F 9/5083 Techniques for rebalancing the load in a distributed system

Abstract

The technology disclosed relates to discovering a previously unknown attribute of stream processing systems according to which client offsets or client subscription queries for a streaming data store rapidly converge to a dynamic tip of a data stream that includes the most recent messages or events. In particular, it relates to grouping clients into bins to reduce a number of queries to the streaming data store by several orders of magnitude when servicing tens, hundreds, thousands or millions of clients. The bin count is further reduced by coalescing bins that have overlapping offsets. It also relates to establishing separate caches only for the current tips of data streams and serving the bins from the caches instead of the backend data store using group queries. Further, the caches are periodically updated to include the most recent messages or events appended to the dynamic tips of the data streams.

Claims

1. A method of reducing a number of queries to a message data store of at least one storage device when servicing a plurality of clients, with each client in the plurality of clients requesting subscription to the data store of the at least one storage device at any available message offset in the data store of the at least one storage device as a starting offset for streaming messages, the method including:

grouping, by at least one processor communicatively coupled to the at least one storage device, client subscription requests of the plurality of clients into one or more bins based on respective subscription offsets identified in the requests of the plurality of clients and respective current bin offset ranges of the one or more bins; and

reducing, by the at least one processor, queries to the data store of the at least one storage device when streaming messages to the client subscription requests of the plurality of clients starting from their respective subscription offsets by issuing against the data store a single block query for each of the bins instead of issuing individual queries for each of the client subscription requests.

2. The method of claim 1, wherein the respective current bin offset ranges of the one or more bins represent lower and upper logical boundaries of a segment of a particular message sequence currently maintained by the data store.

3. The method of claim 2, wherein a start offset of a new bin'"'"'s bin offset range is selected based on a subscription offset of at least one client subscription request in the new bin.

4. The method of claim 2, wherein a start offset of a particular bin'"'"'s bin offset range is determined based on a latest message offset streamed to at least one of the client subscription requests in the particular bin.

5. The method of claim 2, wherein an end offset of each bin'"'"'s bin offset range is determined based on a preconfigured bin size measured in number of messages.

6. The method of claim 1, wherein the grouping of the client subscription requests of the plurality of clients comprises:

receiving, at the at least one processor, the client subscription requests from the plurality of clients, wherein the subscription requests identify as subscription offsets the clients'"'"' respective starting offsets for streaming messages from the data store, and wherein the subscription offsets are distributed across various message offsets currently or historically maintained by the data store;

creating, at the at least one processor, the one or more bins to group the client subscription requests of the plurality of clients based on the distribution of the subscription offsets and a preconfigured bin size, wherein each of the bins has a bin offset range with a start offset selected based on one or more subscription offsets and an end offset selected based on the bin size,

and

wherein the reducing the queries to the data store comprises responding to the client subscription requests by issuing against the data store a single block query for each of the bins instead of issuing individual queries for each of the client subscription requests.

7. The method of claim 1, further including:

for a particular block query issued for a particular bin with a set of the client subscription requests of the plurality of clients, retrieving a message batch from the data store, wherein the message batch has a message offset range with a start offset and an end offset;

streaming the retrieved message batch to the particular bin'"'"'s set of the client subscription requests of the plurality of clients;

advancing the particular bin by updating the particular bin'"'"'s bin offset range based on the end offset of the message batch'"'"'s message offset range and a preconfigured bin size; and

regrouping the set of the of the client subscription requests of the plurality of clients in the advanced particular bin.

8. The method of claim 1, further including:

determining that message offsets of messages streamed to the client subscription requests of the plurality of clients have respectively evolved from a varied distribution across the data store to a cumulative convergence towards a tip message offset in the data store, wherein the tip message offset identifies a latest message in the data store;

regrouping the client subscription requests of the plurality of clients, previously grouped into different bins, into a tip bin with a bin offset range that includes the tip message offset; and

servicing the client subscription requests of the plurality of clients by issuing against the data store a single block query for the tip bin instead of issuing multiple block queries for the different bins.

9. The method of claim 1, further including:

using multiple worker threads to simultaneously process each of the bins.

10. The method of claim 9, further including:

dispatching the bins to the worker threads, including;

during execution, comparing a count of available worker threads against a number of the bins;

when a count of available worker threads equals or exceeds the number of bins, concurrently processing the bins at the available worker threads; and

when there are fewer available worker threads than the number of bins, multiplexing the bins sequentially over the available worker threads.

11. The method of claim 1, wherein the grouping of client subscription requests of the plurality of clients into the one or more bins comprises grouping the client subscription requests of a particular bin into session batches based on a preconfigured session batch size;

and

concurrently dispatching multiple session batches to multiple worker threads for simultaneous processing, wherein the dispatching follows a natural ordering of respective message offsets of the session batches.

12. The method of claim 11, further including:

processing a particular session batch using at a first worker thread;

forking the first working thread to generate one or more new worker threads, wherein the first and new worker threads share a thread pool and utilize a fork-join framework; and

simultaneously streaming message batches to each client subscription request of one of the plurality of clients in the particular session batch using the generated new worker threads.

13. The method of claim 1, further including:

handling a speedy bin that is overtaking a more current bin with more current message offsets than the speedy bin by coalescing the speedy bin with the more current bin, thereby further reducing block queries to the data store,

wherein the speedy bin has an end offset that is greater that a start offset of the current bin.

14. The method of claim 13, further including:

regrouping client subscription requests of the plurality of clients from the speedy bin into the more current bin as session batches.

15. The method of claim 1, further including:

handling a speedy bin that is overtaking a more current bin with more current message offsets than the speedy bin by coalescing the more current bin with the speedy bin, thereby further reducing block queries to the data store,

wherein the speedy bin has an end offset that is greater that a start offset of the more current bin.

16. The method of claim 15, further including:

regrouping client subscription requests of the plurality of clients from the more current bin into the speedy bin as session batches.

17. The method of claim 1, further including:

handling a slow client subscription request of one of the plurality of clients that has a blocked transport queue by setting aside service of the slow client subscription request of one of the plurality of clients until its transport queue is ready to receive additional messages, then reassigning the slow client subscription request of one of the plurality of clients to a bin of the one or more bins based on the current message offset of the slow client subscription request of one of the plurality of clients.

18. The method of claim 1, further including:

detecting an existing client subscription request of one of the plurality of clients dropping out of a particular bin of the plurality of bins;

checking current message offsets of other client subscription requests of the plurality of clients remaining in the particular bin; and

in response to determining that either a current actual start offset or a current actual end offset of the particular bin is within a more current bin'"'"'s current bin offset range, coalescing the particular bin with the more current bin.

19. A non-transitory computer readable storage medium including computer program instructions that, when executed on a processor, implement a method comprising:

receiving selections, at the processor, for a plurality of client subscription requests for available service offsets in an unbounded data stream from a data store as starting offsets for streaming services to the respective client subscription requests;

binning, by the processor, the client subscription requests in bins based on the respective selected starting offsets in the data stream and current offsets of the bins for the client subscription requests;

receiving, at the data store, at least one block query against the data stream based on the binned clients; and

accessing data, by the processor, returned by the at least one block query for clients in a particular bin using a plurality of worker threads.

20. A system including one or more processors coupled to memory, the memory including computer instructions that, when executed on the one or more processors, implement actions comprising:

receiving selections, at the one or more processor, for a plurality of client subscription requests for available service offsets in an unbounded data stream from a data store as starting offsets for streaming services to the respective client subscription requests;

binning, at the one or more processors, the client subscription services in bins based on the respective selected starting offsets in the data stream and current offsets of the bins for the client subscription requests;

receiving, at the data store, at least one block query against the data stream based on the binned clients; and

accessing data, by the processor, returned by the at least one block query for clients in a particular bin using a plurality of worker threads.