Systems and methods for implementing bulk handling in asynchronous processing

• US 10,019,297 B2
• Filed: 06/16/2014
• Issued: 07/10/2018
• Est. Priority Date: 04/03/2013
• Status: Active Grant

• Current Assignee: Salesforce.com, Inc.
• Original Assignee: Salesforce.com, Inc.
• Inventors: Helmich, Marco, Yadav, Ananya, Murugesan, Praveen, Devadhar, Vijayanth
• Primary Examiner(s): Lu, Charles
• Application Number:
  US14/306,113
• Publication Number:
  US 20140304246A1
• Time in Patent Office: 1,485 Days
• CPC Class Codes: G06F 16/2455 Query execution
• CPC Class Codes: G06F 2209/548 Queue
• CPC Class Codes: G06F 9/546 Message passing systems or structures, e.g. queues
• CPC Class Codes: H04L 67/55 Push-based network services

Abstract

Systems and methods for implementing bulk handling in asynchronous processing are described. For example, an exemplary system includes a processor and a memory to execute instructions at the system; a broker to enqueue received messages; a push connection from the broker to a thread, in which the broker is to push an initial message to the thread for processing; a pull connection from the thread to the broker, in which the thread is to request one or more additional messages for bulk processing with the initial message from the broker; a query interface to issue a database query from the thread to a data store, the database query specifying the initial message and the one or more additional messages as a set of messages to obtain a lock for the set of messages; and a message processing engine to handle the set of messages in bulk based on whether the lock for the set of messages is obtained.

Claims

1. A system to execute within a host organization, wherein the system comprises:

a processor and a memory;

wherein the system executes instructions via the processor and memory which cause the system to perform the following operations;

receiving messages at a message broker;

enqueuing the received messages within a message queue of the message broker;

establishing a push connection from the broker to a thread and pushing an initial message from the message queue to the thread and processing the initial message via the thread;

establishing a pull connection from the thread to the broker and requesting one or more additional messages from the message queue, and bulk processing the one or more additional messages with the initial message from the broker;

issuing a database query via a query interface by issuing the database query from the thread to a data store, wherein the database query specifies the initial message and the one or more additional messages as a set of messages and obtaining a lock for the set of messages; and

handling, via a message processing engine, by processing the set of messages in bulk when the lock for the set of messages is obtained and re-enqueuing any of the set of messages for which the lock is not obtained.

2. The system of claim 1:

wherein the lock for the set of messages is obtained responsive to the database query; and

wherein the set of messages is processed by the message processing engine concurrently in bulk.

3. The system of claim 2, wherein subsequent to successful processing of the set of messages concurrently in bulk by the message processing engine the query interface deletes records corresponding to the set of messages from the data store and commits the delete.

4. The system of claim 1, wherein the data store comprises a relational database;

wherein the relational database stores records corresponding to the messages received and enqueued by the broker.

5. The system of claim 4:

wherein issuing the database query from the query interface to the data store further comprises issuing an update via the database query specifying the initial message and the one or more additional messages as the set of messages and obtaining the lock for the set of messages; and

wherein the lock for the set of message is obtained responsive to the database query when a record exists in the relational database for each of the messages in the set and additionally when the record is not already subject to a lock.

6. The system of claim 1:

wherein the query interface issues a second database query from the thread to the data store responsive to the lock for the set of messages failing to be obtained, wherein the second database query specifies records corresponding to each of the messages in the set via a SELECT statement within an SQL query; and

wherein the data store returns a record set responsive to the SELECT statement listing all records in the data store corresponding to the messages in the set that presently exist within the data store regardless of whether or not the records are locked.

7. The system of claim 6, wherein the message processing engine further determines which records corresponding to the set of messages are non-existent in the data store as having been previously deleted based on a differential comparison between the messages in the set and the record set returned responsive to the SELECT statement.

8. The system of claim 1:

wherein bulk processing the one or more additional messages with the initial message from the broker includes database transactions processed against a first database within the host organization, the first database storing customer data; and

wherein the data store comprises a second database storing records corresponding to the messages received and enqueued by the broker.

9. The system of claim 8:

wherein the first database within the host organization storing customer data comprises a multi-tenant database system of the host organization storing customer data on behalf of a plurality of separate and distinct customer organizations; and

wherein the second database comprises a relational database operated within the host organization stores the records corresponding to the messages received and having been enqueued by the broker without being accessible to any of the plurality of separate and distinct customer organizations which utilize the multi-tenant database system.

10. The system of claim 9, wherein each of the plurality of separate and distinct customer organizations which utilize the multi-tenant database system is an entity selected from the group consisting of:

a separate and distinct remote organization, an organizational group within the host organization, a business partner of the host organization, or a customer organization that subscribes to cloud computing services provided by the host organization.

11. The system of claim 1, wherein the system further comprises the message queue enqueuing the received messages at the direction of the broker.

12. The system of claim 1, wherein processing the initial message via the thread includes the initial message comprising comprises a flag indicating application of bulk processing of the initial message concurrently with other enqueued messages.

13. The system of claim 1, wherein processing the initial message via the thread includes the initial message comprising information specifying other enqueued messages are subject to bulk processing concurrently with the initial message, wherein the information comprises at least one of:

a bulk processing flag;

an enumerated set of additional message IDs to be processed concurrently with the initial message;

a message type for which additional messages having the same message type are to be processed in bulk concurrently with the initial message;

a range of message IDs to be processed concurrently with the initial message;

search criteria to locate one or more additional messages to be processed concurrently with the initial message; and

a quantity or numerical quantity range of additional messages to be processed concurrently with the initial message.

14. The system of claim 1, wherein the message processing engine analyzes the initial message for a flag, count, search criteria, enumerated set, or message type and subsequently retrieves the one or more additional messages from the broker and bulk processing the subsequently retrieved one or more additional messages concurrently with the initial message.

15. The system of claim 1, further comprising determining, via the message processing engine the lock for the set of messages is not obtained;

issuing, via the query interface a second database query requesting a record set corresponding to the set of messages;

wherein re-enqueuing any of the set of messages for which the lock is not obtained comprises the message processing engine re-enqueuing any message of the set determined to exist within the data store but for which the lock cannot be obtained; and

wherein the method further comprises discarding, via the message processing engine any message of the set determined to no longer exist as a record within the data store.

16. The system of claim 1:

wherein any message for which a lock is obtained pursuant to the first database query indicates a record corresponding to the message exists in the data store and is not actively undergoing processing by any thread;

wherein any message for which a lock is not obtained pursuant to the first database query and which is determined to exist within the data store pursuant to the second database query indicates a record corresponding to the message exists in the data store but is actively undergoing processing by another thread or processing of the message by another thread has terminated abnormally and will be handled by future processing after re-enqueuing of the corresponding message with the broker; and

wherein any message for which a lock is not obtained pursuant to the first database query and which is determined to be non-existent exist within the data store pursuant to the second database query indicates processing of the message by another thread successfully processed and deleted the corresponding record from the data store and the message may be permissibly discarded as an erroneous duplicate message.

17. The system of claim 1:

wherein the thread constitutes one thread among a plurality of threads available for processing the received messages on behalf of the host organization and processing at least the one thread available for processing; and

wherein each thread is one of a server in a pool of servers, a computing blade amongst a group of blades, a single computer node within a computing grid, a processing work thread executing via processor cores amongst a plurality of processing work threads, or one of a plurality of computing nodes amongst a distributed network of computing nodes.

18. A method executing within a system of a host organization, the system having a processor and a memory therein executing instructions within the system, wherein the method comprises:

receiving messages at a message broker;

enqueuing the received messages within a message queue of the message broker;

establishing a push connection from the broker to a thread and pushing an initial message from the message queue to the thread and processing the initial message via the thread;

establishing a pull connection from the thread to the broker and requesting one or more additional messages from the message queue, and bulk processing the one or more additional messages with the initial message from the broker;

issuing a database query via a query interface by issuing the database query from the thread to a data store, wherein the database query specifies the initial message and the one or more additional messages as a set of messages and obtaining a lock for the set of messages; and

handling, via a message processing engine, by processing the set of messages in bulk when the lock for the set of messages is obtained and re-enqueuing any of the set of messages for which the lock is not obtained.

19. The method of claim 18, further comprising:

obtaining the lock for the set of messages responsive to the database query;

processing the set of messages concurrently in bulk; and

determining processing of the set of messages was successful.

20. The method of claim 19, further comprising:

de-queuing the set of messages from the message queue of the message broker; and

issuing a new database query to the data store to delete all records corresponding to the set of messages successfully processed and committing the delete.

21. The method of claim 18:

wherein the data store comprises a relational database storing records corresponding to the messages received and enqueued by the broker;

wherein the method further comprises obtaining the lock for the set of messages by issuing an update via the database query specifying the initial message and the one or more additional messages as the set of messages; and

wherein the lock for the set of message is obtained responsive to the database query when a record exists in the relational database for each of the messages in the set and when the record is not already subject to a lock.

22. The method of claim 18, further comprising:

issuing a second database query to the data store responsive to the lock for the set of messages failing to be obtained, wherein the second database query specifies records corresponding to each of the messages in the set via a SELECT statement within an SQL query; and

receiving a record set from the data store responsive to the SELECT statement listing all records in the data store corresponding to the messages in the set that presently exist within the data store regardless of whether or not the records are locked.

23. The method of claim 22, further comprising:

determining which records corresponding to the set of messages are non-existent in the data store as having been previously deleted based on a differential comparison between the messages in the set and the record set returned responsive to the SELECT statement;

wherein the initial message and the one or more additional messages for bulk processing with the initial message embody database transactions to be processed against a first database within the host organization, the first database storing customer data; and

wherein the data store comprises a second database to store records corresponding to the messages received and enqueued by the broker.

24. Non-transitory computer readable storage media having instructions stored thereon that, when executed by a processor of a system, the instructions cause the system to perform operations comprising:

receiving messages at a message broker;

enqueuing the received messages within a message queue of the message broker;

establishing a push connection from the broker to a thread and pushing an initial message from the message queue to the thread and processing the initial message via the thread;

establishing a pull connection from the thread to the broker and requesting one or more additional messages from the message queue, and bulk processing the one or more additional messages with the initial message from the broker;

issuing a database query via a query interface by issuing the database query from the thread to a data store, wherein the database query specifies the initial message and the one or more additional messages as a set of messages and obtaining a lock for the set of messages; and

handling, via a message processing engine, by processing the set of messages in bulk when the lock for the set of messages is obtained and re-enqueuing any of the set of messages for which the lock is not obtained.

25. The non-transitory computer readable storage media of claim 24, further comprising:

issuing a second database query to the data store responsive to the lock for the set of messages not being obtained, wherein the second database query specifies records corresponding to each of the messages in the set via a SELECT statement within an SQL query;

receiving a record set from the data store responsive to the SELECT statement listing all records in the data store corresponding to the messages in the set that presently exist within the data store regardless of whether or not the records are locked;

determining which records corresponding to the set of messages are non-existent in the data store as having been previously deleted based on a differential comparison between the messages in the set and the record set returned responsive to the SELECT statement;

wherein the initial message and the one or more additional messages for bulk processing with the initial message embody database transactions to be processed against a first database within the host organization, the first database storing customer data; and

wherein the data store comprises a second database to store records corresponding to the messages received and enqueued by the broker.