Systems and methods for storing message data
Methods, systems, and apparatus, including computer programs encoded on a computer storage medium, are described for storing message data in a PubSub system. In certain examples, messages are received from a plurality of publishers for a plurality of distinct channels. The messages are ordered and stored in a plurality of buffers, with each channel having its own respective buffer. After a message has been written to a writable portion of the buffer for a channel, a pointer demarking a boundary between a readable portion of the buffer and the writeable portion of the buffer is advanced in an atomic operation. Following the atomic operation, the message resides in the readable portion and may be accessed by PubSub system components and/or processes. In general, one or more subscribers, components, or processes may read messages from the readable portion, in parallel.
1. A method comprising:
performing by one or more computers:
receiving from a plurality of publishers a plurality of messages, each of the messages being associated with one of a plurality of distinct channels;
ordering the messages associated with each channel;
storing each message of each of the channels in a respective buffer for the channel according to the order of the messages assigned to the channel, wherein storing comprises storing the message in a writable portion of the buffer and advancing a pointer demarking a boundary between a readable portion of the buffer and the writeable portion of the buffer in an atomic operation such that the message is in the readable portion of the buffer after the atomic operation has completed; and
allowing one or more subscribers to read from the readable portion of one or more of the buffers during the storing.
2. The method of claim 1 wherein the atomic operation cannot be interrupted by another process or thread of execution.
3. The method of claim 1 wherein storing the message in the writable portion of the buffer comprises storing a length of the message at a first location in the writable portion of the buffer and storing the message in the writable portion of the buffer following the first location.
4. The method of claim 1 wherein advancing the pointer demarking the boundary between the readable portion of the buffer and the writable portion of the buffer in an atomic operation comprises storing a sum of the length of the message and a current value of the pointer in the pointer.
5. The method of claim 1 wherein each buffer only stores messages for a single channel.
6. The method of claim 1 wherein each buffer for a particular channel expires at a different time based on the time-to-live for the buffer.
7. The method of claim 1 wherein a particular buffer corresponds to a writing process on one of the computers.
8. The method of claim 1 wherein each buffer has a respective time-to-live upon expiration of which will cause the buffer to be inaccessible to publishers and subscribers.
9. The method of claim 1 wherein ordering the messages associated with each channel comprises ordering the messages according to respective receipt times of the messages.
10. A system comprising:
a non-transitory computer readable medium having instructions stored thereon; and
a data processing apparatus configured to execute the instructions to perform operations comprising:
receiving from a plurality of publishers a plurality of messages, each of the messages being associated with one of a plurality of distinct channels;
ordering the messages associated with each channel;
storing each message of each of the channels in a respective buffer for the channel according to the order of the messages assigned to the channel, wherein storing comprises storing the message in a writable portion of the buffer and advancing a pointer demarking a boundary between a readable portion of the buffer and the writeable portion of the buffer in an atomic operation such that the message is in the readable portion of the buffer after the atomic operation has completed; and
allowing one or more subscribers to read from the readable portion of one or more of the buffers during the storing.
11. The system of claim 10 wherein the atomic operation cannot be interrupted by another process or thread of execution.
12. The system of claim 10 wherein storing the message in the writable portion of the buffer comprises storing a length of the message at a first location in the writable portion of the buffer and storing the message in the writable portion of the buffer following the first location.
13. The system of claim 10 wherein advancing the pointer demarking the boundary between the readable portion of the buffer and the writable portion of the buffer in an atomic operation comprises storing a sum of the length of the message and a current value of the pointer in the pointer.
14. The system of claim 10 wherein each buffer only stores messages for a single channel.
15. The system of claim 10 wherein each buffer for a particular channel expires at a different time based on the time-to-live for the buffer.
16. The system of claim 10 wherein a particular buffer corresponds to a writing process on one of the computers.
17. The system of claim 10 wherein each buffer has a respective time-to-live upon expiration of which will cause the buffer to be inaccessible to publishers and subscribers.
18. The system of claim 10 wherein ordering the messages associated with each channel comprises ordering the messages according to respective receipt times of the messages.
19. A computer program product stored in one or more non-transitory storage media for controlling a processing mode of a data processing apparatus, the computer program product being executable by the data processing apparatus to cause the data processing apparatus to perform operations comprising:
receiving from a plurality of publishers a plurality of messages, each of the messages being associated with one of a plurality of distinct channels;
ordering the messages associated with each channel;
storing each message of each of the channels in a respective buffer for the channel according to the order of the messages assigned to the channel, wherein storing comprises storing the message in a writable portion of the buffer and advancing a pointer demarking a boundary between a readable portion of the buffer and the writeable portion of the buffer in an atomic operation such that the message is in the readable portion of the buffer after the atomic operation has completed; and
allowing one or more subscribers to read from the readable portion of one or more of the buffers during the storing.
20. The computer program product of claim 19 wherein the atomic operation cannot be interrupted by another process or thread of execution.
21. The computer program product of claim 19 wherein storing the message in the writable portion of the buffer comprises storing a length of the message at a first location in the writable portion of the buffer and storing the message in the writable portion of the buffer following the first location.
22. The computer program product of claim 19 wherein advancing the pointer demarking the boundary between the readable portion of the buffer and the writable portion of the buffer in an atomic operation comprises storing a sum of the length of the message and a current value of the pointer in the pointer.
23. The computer program product of claim 19 wherein each buffer only stores messages for a single channel.
24. The computer program product of claim 19 wherein each buffer for a particular channel expires at a different time based on the time-to-live for the buffer.
25. The computer program product of claim 19 wherein a particular buffer corresponds to a writing process on one of the computers.
26. The computer program product of claim 19 wherein each buffer has a respective time-to-live upon expiration of which will cause the buffer to be inaccessible to publishers and subscribers.
27. The computer program product of claim 19 wherein ordering the messages associated with each channel comprises ordering the messages according to respective receipt times of the messages.