Chapman: Building a High-Performance Distributed Task Service with MongoDB

When you're building a web application, you want to respond to every request as quickly as possible. The usual approach is to use an asynchronous job queue like Sidekiq, Resque, Celery, RQ, or a number of other frameworks to handle those tasks outside the request/response cycle in a separate 'worker' process. Unfortunately, many of these frameworks either require the deployment of Redis, RabbitMQ, or some other request broker, or they resort to polling a database for new work to do. Chapman is a distributed task queue built on MongoDB that avoids gratuitous polling, using tailable cursors with the oplog to provide notifications of incoming work. Inspired by Celery, Chapman also supports task graphs, where multiple tasks that depend on each other can be executed by the system asynchronously. Come learn how Synapp.io is using MongoDB and Chapman to handle its core data processing needs.

Transcript

1. Chapman: Building a Distributed Job Queue in MongoDB Rick Copeland @rick446 @synappio rick@synapp.io
2. @rick446 @synappio Getting to Know One Another
3. @rick446 @synappio Getting to Know One Another Rick
4. @rick446 @synappio Getting to Know One Another Rick
5. @rick446 @synappio What You’ll Learn
6. @rick446 @synappio What You’ll Learn How to…
7. @rick446 @synappio What You’ll Learn How to… Build a task queue in MongoDB
8. @rick446 @synappio What You’ll Learn How to… Build a task queue in MongoDB
9. @rick446 @synappio What You’ll Learn How to… Build a task queue in MongoDB Bring consistency to distributed systems (without transactions)
10. @rick446 @synappio What You’ll Learn How to… Build a task queue in MongoDB Bring consistency to distributed systems (without transactions)
11. @rick446 @synappio What You’ll Learn How to… Build a task queue in MongoDB Bring consistency to distributed systems (without transactions) Build low-latency reactive systems
12. @rick446 @synappio Why a Queue?
13. @rick446 @synappio Why a Queue? • Long-running task (or longer than the web can wait)
14. @rick446 @synappio Why a Queue? • Long-running task (or longer than the web can wait) • Farm out chunks of work for performance
15. @rick446 @synappio Things I Worry About
16. @rick446 @synappio Things I Worry About • Priority
17. @rick446 @synappio Things I Worry About • Priority • Latency
18. @rick446 @synappio Things I Worry About • Priority • Latency • Unreliable workers
19. @rick446 @synappio Queue Options
20. @rick446 @synappio Queue Options • SQS? No priority
21. @rick446 @synappio Queue Options • SQS? No priority • Redis? Can’t overflow memory
22. @rick446 @synappio Queue Options • SQS? No priority • Redis? Can’t overflow memory • Rabbit-MQ? Lack of visibility
23. @rick446 @synappio Queue Options • SQS? No priority • Redis? Can’t overflow memory • Rabbit-MQ? Lack of visibility • ZeroMQ? Lack of persistence
24. @rick446 @synappio Queue Options • SQS? No priority • Redis? Can’t overflow memory • Rabbit-MQ? Lack of visibility • ZeroMQ? Lack of persistence • What about MongoDB?
25. @rick446 @synappio Chapman Graham Arthur Chapman 8 January 1941 – 4 October 1989
26. @rick446 @synappio Roadmap
27. @rick446 @synappio Roadmap • Building a scheduled priority queue
28. @rick446 @synappio Roadmap • Building a scheduled priority queue • Handling unreliable workers
29. @rick446 @synappio Roadmap • Building a scheduled priority queue • Handling unreliable workers • Shared resources
30. @rick446 @synappio Roadmap • Building a scheduled priority queue • Handling unreliable workers • Shared resources • Managing Latency
31. @rick446 @synappio Building a Scheduled Priority Queue
32. @rick446 @synappio Step 1: Simple Queue db.message.insert({! "_id" : NumberLong("3784707300388732067"),! "data" : BinData(...),! "s" : {! "status" : "ready",! "ts_enqueue" : ISODate("2015-03-02T15:27:29.228Z")! }! });! ! db.message.ensureIndex({'s.status': 1, 's.ts_enqueue': 1});! ! db.runCommand(! {! findAndModify: "message",! query: { 's.status': 'ready' },! sort: {'s.ts_enqueue': 1},! update: { '$set': {'s.status': 'reserved'} },! }! );!
33. @rick446 @synappio Step 1: Simple Queue db.message.insert({! "_id" : NumberLong("3784707300388732067"),! "data" : BinData(...),! "s" : {! "status" : "ready",! "ts_enqueue" : ISODate("2015-03-02T15:27:29.228Z")! }! });! ! db.message.ensureIndex({'s.status': 1, 's.ts_enqueue': 1});! ! db.runCommand(! {! findAndModify: "message",! query: { 's.status': 'ready' },! sort: {'s.ts_enqueue': 1},! update: { '$set': {'s.status': 'reserved'} },! }! );! FIFO
34. @rick446 @synappio Step 1: Simple Queue db.message.insert({! "_id" : NumberLong("3784707300388732067"),! "data" : BinData(...),! "s" : {! "status" : "ready",! "ts_enqueue" : ISODate("2015-03-02T15:27:29.228Z")! }! });! ! db.message.ensureIndex({'s.status': 1, 's.ts_enqueue': 1});! ! db.runCommand(! {! findAndModify: "message",! query: { 's.status': 'ready' },! sort: {'s.ts_enqueue': 1},! update: { '$set': {'s.status': 'reserved'} },! }! );! FIFO Get earliest message for processing
35. @rick446 @synappio Step 1: Simple Queue
36. @rick446 @synappio Step 1: Simple Queue Good
37. @rick446 @synappio Step 1: Simple Queue Good • Guaranteed FIFO
38. @rick446 @synappio Step 1: Simple Queue Good • Guaranteed FIFO Bad
39. @rick446 @synappio Step 1: Simple Queue Good • Guaranteed FIFO Bad • No priority (other than FIFO)
40. @rick446 @synappio Step 1: Simple Queue Good • Guaranteed FIFO Bad • No priority (other than FIFO) • No handling of worker problems
41. @rick446 @synappio Step 2: Scheduled Messages db.message.insert({! "_id" : NumberLong("3784707300388732067"),! "data" : BinData(...),! "s" : {! "status" : "ready",! “ts_after" : ISODate(…),! "ts_enqueue" : ISODate("2015-03-02T15:27:29.228Z")! }! });! ! db.message.ensureIndex(! {'s.status': 1, 's.ts_enqueue': 1});! ! db.runCommand(! {! findAndModify: "message",! query: { 's.status': 'ready', ’s.ts_after': {$lt: now }},! sort: {'s.ts_enqueue': 1},! update: { '$set': {'s.status': 'reserved'} },! }! );
42. @rick446 @synappio Step 2: Scheduled Messages db.message.insert({! "_id" : NumberLong("3784707300388732067"),! "data" : BinData(...),! "s" : {! "status" : "ready",! “ts_after" : ISODate(…),! "ts_enqueue" : ISODate("2015-03-02T15:27:29.228Z")! }! });! ! db.message.ensureIndex(! {'s.status': 1, 's.ts_enqueue': 1});! ! db.runCommand(! {! findAndModify: "message",! query: { 's.status': 'ready', ’s.ts_after': {$lt: now }},! sort: {'s.ts_enqueue': 1},! update: { '$set': {'s.status': 'reserved'} },! }! ); Min Valid Time
43. @rick446 @synappio Step 2: Scheduled Messages db.message.insert({! "_id" : NumberLong("3784707300388732067"),! "data" : BinData(...),! "s" : {! "status" : "ready",! “ts_after" : ISODate(…),! "ts_enqueue" : ISODate("2015-03-02T15:27:29.228Z")! }! });! ! db.message.ensureIndex(! {'s.status': 1, 's.ts_enqueue': 1});! ! db.runCommand(! {! findAndModify: "message",! query: { 's.status': 'ready', ’s.ts_after': {$lt: now }},! sort: {'s.ts_enqueue': 1},! update: { '$set': {'s.status': 'reserved'} },! }! ); Min Valid Time Get earliest message for processing
44. @rick446 @synappio Step 2: Scheduled Messages
45. @rick446 @synappio Step 2: Scheduled Messages Good
46. @rick446 @synappio Step 2: Scheduled Messages Good • Easy to build periodic tasks
47. @rick446 @synappio Step 2: Scheduled Messages Good • Easy to build periodic tasks Bad
48. @rick446 @synappio Step 2: Scheduled Messages Good • Easy to build periodic tasks Bad • Be careful with the word “now”
49. @rick446 @synappio Step 3: Priority db.message.insert({! "_id" : NumberLong("3784707300388732067"),! "data" : BinData(...),! "s" : {! "status" : "ready",! "pri": 30128,! "ts_enqueue" : ISODate("2015-03-02T15:27:29.228Z")! }! });! ! db.message.ensureIndex({'s.status': 1, 's.pri': -1, 's.ts_enqueue': 1});! ! db.runCommand(! {! findAndModify: "message",! query: { 's.status': 'ready' },! sort: {'s.pri': -1, 's.ts_enqueue': 1},! update: { '$set': {'s.status': 'reserved'} },! }! );
50. @rick446 @synappio Step 3: Priority db.message.insert({! "_id" : NumberLong("3784707300388732067"),! "data" : BinData(...),! "s" : {! "status" : "ready",! "pri": 30128,! "ts_enqueue" : ISODate("2015-03-02T15:27:29.228Z")! }! });! ! db.message.ensureIndex({'s.status': 1, 's.pri': -1, 's.ts_enqueue': 1});! ! db.runCommand(! {! findAndModify: "message",! query: { 's.status': 'ready' },! sort: {'s.pri': -1, 's.ts_enqueue': 1},! update: { '$set': {'s.status': 'reserved'} },! }! ); Add Priority
51. @rick446 @synappio Step 3: Priority db.message.insert({! "_id" : NumberLong("3784707300388732067"),! "data" : BinData(...),! "s" : {! "status" : "ready",! "pri": 30128,! "ts_enqueue" : ISODate("2015-03-02T15:27:29.228Z")! }! });! ! db.message.ensureIndex({'s.status': 1, 's.pri': -1, 's.ts_enqueue': 1});! ! db.runCommand(! {! findAndModify: "message",! query: { 's.status': 'ready' },! sort: {'s.pri': -1, 's.ts_enqueue': 1},! update: { '$set': {'s.status': 'reserved'} },! }! ); Add Priority
52. @rick446 @synappio Step 3: Priority
53. @rick446 @synappio Step 3: Priority Good
54. @rick446 @synappio Step 3: Priority Good • Priorities are handled
55. @rick446 @synappio Step 3: Priority Good • Priorities are handled • Guaranteed FIFO within a priority
56. @rick446 @synappio Step 3: Priority Good • Priorities are handled • Guaranteed FIFO within a priority Bad
57. @rick446 @synappio Step 3: Priority Good • Priorities are handled • Guaranteed FIFO within a priority Bad • No handling of worker problems
58. @rick446 @synappio Handling Unreliable Workers
59. @rick446 @synappio Approach 1 Timeouts db.message.insert({! "_id" : NumberLong("3784707300388732067"),! "data" : BinData(...),! "s" : {! "status" : "ready",! "pri": 30128,! "ts_enqueue" : ISODate("2015-03-02T15:27:29.228Z"),! "ts_timeout" : ISODate("2025-01-01T00:00:00.000Z")! }! });! ! db.message.ensureIndex({“s.status": 1, “s.ts_timeout": 1})! !
60. @rick446 @synappio Approach 1 Timeouts db.message.insert({! "_id" : NumberLong("3784707300388732067"),! "data" : BinData(...),! "s" : {! "status" : "ready",! "pri": 30128,! "ts_enqueue" : ISODate("2015-03-02T15:27:29.228Z"),! "ts_timeout" : ISODate("2025-01-01T00:00:00.000Z")! }! });! ! db.message.ensureIndex({“s.status": 1, “s.ts_timeout": 1})! ! Far-future placeholder
61. @rick446 @synappio // Reserve message! db.runCommand(! {! findAndModify: "message",! query: { 's.status': 'ready' },! sort: {'s.pri': -1, 's.ts_enqueue': 1},! update: { '$set': {! 's.status': 'reserved',! 's.ts_timeout': now + processing_time } }! }! );! ! // Timeout message ("unlock")! db.message.update(! {'s.ts_status': 'reserved', 's.ts_timeout': {'$lt': now}},! {'$set': {'s.status': 'ready'}},! {'multi': true}); Approach 1 Timeouts
62. @rick446 @synappio // Reserve message! db.runCommand(! {! findAndModify: "message",! query: { 's.status': 'ready' },! sort: {'s.pri': -1, 's.ts_enqueue': 1},! update: { '$set': {! 's.status': 'reserved',! 's.ts_timeout': now + processing_time } }! }! );! ! // Timeout message ("unlock")! db.message.update(! {'s.ts_status': 'reserved', 's.ts_timeout': {'$lt': now}},! {'$set': {'s.status': 'ready'}},! {'multi': true}); Client sets timeout Approach 1 Timeouts
63. @rick446 @synappio Approach 1 Timeouts
64. @rick446 @synappio Approach 1 Timeouts Good
65. @rick446 @synappio Approach 1 Timeouts Good • Worker failure handled via timeout
66. @rick446 @synappio Approach 1 Timeouts Good • Worker failure handled via timeout Bad
67. @rick446 @synappio Approach 1 Timeouts Good • Worker failure handled via timeout Bad • Requires periodic “unlock” task
68. @rick446 @synappio Approach 1 Timeouts Good • Worker failure handled via timeout Bad • Requires periodic “unlock” task • Slow (but “live”) workers can cause spurious timeouts
69. @rick446 @synappio db.message.insert({! "_id" : NumberLong("3784707300388732067"),! "data" : BinData(...),! "s" : {! "status" : "ready",! "pri": 30128,! "cli": "--------------------------"! "ts_enqueue" : ISODate("2015-03-02T..."),! "ts_timeout" : ISODate("2025-...")! }! }); Approach 2 Worker Identity
70. @rick446 @synappio db.message.insert({! "_id" : NumberLong("3784707300388732067"),! "data" : BinData(...),! "s" : {! "status" : "ready",! "pri": 30128,! "cli": "--------------------------"! "ts_enqueue" : ISODate("2015-03-02T..."),! "ts_timeout" : ISODate("2025-...")! }! }); Client / worker placeholder Approach 2 Worker Identity
71. @rick446 @synappio // Reserve message! db.runCommand(! {! findAndModify: "message",! query: { 's.status': 'ready' },! sort: {'s.pri': -1, 's.ts_enqueue': 1},! update: { '$set': {! 's.status': 'reserved',! 's.cli': ‘client_name:pid',! 's.ts_timeout': now + processing_time } }! }! );! ! // Unlock “dead” client messages! db.message.update(! {'s.status': 'reserved', ! 's.cli': {'$nin': active_clients} },! {'$set': {'s.status': 'ready'}},! {'multi': true});! Approach 2 Worker Identity
72. @rick446 @synappio // Reserve message! db.runCommand(! {! findAndModify: "message",! query: { 's.status': 'ready' },! sort: {'s.pri': -1, 's.ts_enqueue': 1},! update: { '$set': {! 's.status': 'reserved',! 's.cli': ‘client_name:pid',! 's.ts_timeout': now + processing_time } }! }! );! ! // Unlock “dead” client messages! db.message.update(! {'s.status': 'reserved', ! 's.cli': {'$nin': active_clients} },! {'$set': {'s.status': 'ready'}},! {'multi': true});! Mark the worker who reserved the message Approach 2 Worker Identity
73. @rick446 @synappio // Reserve message! db.runCommand(! {! findAndModify: "message",! query: { 's.status': 'ready' },! sort: {'s.pri': -1, 's.ts_enqueue': 1},! update: { '$set': {! 's.status': 'reserved',! 's.cli': ‘client_name:pid',! 's.ts_timeout': now + processing_time } }! }! );! ! // Unlock “dead” client messages! db.message.update(! {'s.status': 'reserved', ! 's.cli': {'$nin': active_clients} },! {'$set': {'s.status': 'ready'}},! {'multi': true});! Mark the worker who reserved the message Messages reserved by dead workers are unlocked Approach 2 Worker Identity
74. @rick446 @synappio Approach 2 Worker Identity
75. @rick446 @synappio Approach 2 Worker Identity Good
76. @rick446 @synappio Approach 2 Worker Identity Good • Worker failure handled via out- of-band detection of live workers
77. @rick446 @synappio Approach 2 Worker Identity Good • Worker failure handled via out- of-band detection of live workers • Handles slow workers
78. @rick446 @synappio Approach 2 Worker Identity Good • Worker failure handled via out- of-band detection of live workers • Handles slow workers
79. @rick446 @synappio Approach 2 Worker Identity Good • Worker failure handled via out- of-band detection of live workers • Handles slow workers Bad
80. @rick446 @synappio Approach 2 Worker Identity Good • Worker failure handled via out- of-band detection of live workers • Handles slow workers Bad • Requires periodic “unlock” task
81. @rick446 @synappio Approach 2 Worker Identity Good • Worker failure handled via out- of-band detection of live workers • Handles slow workers Bad • Requires periodic “unlock” task • Unlock updates can be slow
82. @rick446 @synappio Shared Resources
83. @rick446 @synappio Complex Tasks Group check_smtp Analyze Results Update Reports Pipeline
84. @rick446 @synappio Semaphores
85. @rick446 @synappio Semaphores • Some services perform connection- throttling (e.g. Mailchimp)
86. @rick446 @synappio Semaphores • Some services perform connection- throttling (e.g. Mailchimp) • Some services just have a hard time with 144 threads hitting them simultaneously
87. @rick446 @synappio Semaphores • Some services perform connection- throttling (e.g. Mailchimp) • Some services just have a hard time with 144 threads hitting them simultaneously • Need a way to limit our concurrency
88. @rick446 @synappio Semaphores Semaphore Active: msg1, msg2, msg3, … Capacity: 16 Queued: msg17, msg18, msg19, …
89. @rick446 @synappio Semaphores Semaphore Active: msg1, msg2, msg3, … Capacity: 16 Queued: msg17, msg18, msg19, … • Keep active and queued messages in arrays
90. @rick446 @synappio Semaphores Semaphore Active: msg1, msg2, msg3, … Capacity: 16 Queued: msg17, msg18, msg19, … • Keep active and queued messages in arrays • Releasing the semaphore makes queued messages available for dispatch
91. @rick446 @synappio Semaphores Semaphore Active: msg1, msg2, msg3, … Capacity: 16 Queued: msg17, msg18, msg19, … • Keep active and queued messages in arrays • Releasing the semaphore makes queued messages available for dispatch • Use $slice (2.6) to keep arrays the right size
92. @rick446 @synappio Semaphores: Acquire db.semaphore.insert({! '_id': 'semaphore-name',! 'value': 16,! 'active': [],! 'queued': []});! ! def acquire(sem_id, msg_id, sem_size):! sem = db.semaphore.find_and_modify(! {'_id': sem_id},! update={'$push': {! 'active': {! '$each': [msg_id], ! '$slice': sem_size},! 'queued': msg_id}},! new=True)! if msg_id in sem['active']:! db.semaphore.update(! {'_id': 'semaphore-name'},! {'$pull': {'queued': msg_id}})! return True! return False
93. @rick446 @synappio Semaphores: Acquire db.semaphore.insert({! '_id': 'semaphore-name',! 'value': 16,! 'active': [],! 'queued': []});! ! def acquire(sem_id, msg_id, sem_size):! sem = db.semaphore.find_and_modify(! {'_id': sem_id},! update={'$push': {! 'active': {! '$each': [msg_id], ! '$slice': sem_size},! 'queued': msg_id}},! new=True)! if msg_id in sem['active']:! db.semaphore.update(! {'_id': 'semaphore-name'},! {'$pull': {'queued': msg_id}})! return True! return False Pessimistic update
94. @rick446 @synappio Semaphores: Acquire db.semaphore.insert({! '_id': 'semaphore-name',! 'value': 16,! 'active': [],! 'queued': []});! ! def acquire(sem_id, msg_id, sem_size):! sem = db.semaphore.find_and_modify(! {'_id': sem_id},! update={'$push': {! 'active': {! '$each': [msg_id], ! '$slice': sem_size},! 'queued': msg_id}},! new=True)! if msg_id in sem['active']:! db.semaphore.update(! {'_id': 'semaphore-name'},! {'$pull': {'queued': msg_id}})! return True! return False Pessimistic update Compensation
95. @rick446 @synappio Semaphores: Release def release(sem_id, msg_id, sem_size):! sem = db.semaphore.find_and_modify(! {'_id': sem_id},! update={'$pull': {! 'active': msg_id, ! 'queued': msg_id}},! new=True)! ! while len(sem['active']) < sem_size and sem['queued']:! wake_msg_ids = sem['queued'][:sem_size]! updated = self.cls.m.find_and_modify(! {'_id': sem_id},! update={'$pullAll': {'queued': wake_msg_ids}},! new=True)! for msgid in wake_msg_ids:! make_dispatchable(msgid)! sem = updated
96. @rick446 @synappio Semaphores: Release def release(sem_id, msg_id, sem_size):! sem = db.semaphore.find_and_modify(! {'_id': sem_id},! update={'$pull': {! 'active': msg_id, ! 'queued': msg_id}},! new=True)! ! while len(sem['active']) < sem_size and sem['queued']:! wake_msg_ids = sem['queued'][:sem_size]! updated = self.cls.m.find_and_modify(! {'_id': sem_id},! update={'$pullAll': {'queued': wake_msg_ids}},! new=True)! for msgid in wake_msg_ids:! make_dispatchable(msgid)! sem = updated Actually release
97. @rick446 @synappio Semaphores: Release def release(sem_id, msg_id, sem_size):! sem = db.semaphore.find_and_modify(! {'_id': sem_id},! update={'$pull': {! 'active': msg_id, ! 'queued': msg_id}},! new=True)! ! while len(sem['active']) < sem_size and sem['queued']:! wake_msg_ids = sem['queued'][:sem_size]! updated = self.cls.m.find_and_modify(! {'_id': sem_id},! update={'$pullAll': {'queued': wake_msg_ids}},! new=True)! for msgid in wake_msg_ids:! make_dispatchable(msgid)! sem = updated Actually release Awaken queued message(s)
98. @rick446 @synappio Semaphores: Release def release(sem_id, msg_id, sem_size):! sem = db.semaphore.find_and_modify(! {'_id': sem_id},! update={'$pull': {! 'active': msg_id, ! 'queued': msg_id}},! new=True)! ! while len(sem['active']) < sem_size and sem['queued']:! wake_msg_ids = sem['queued'][:sem_size]! updated = self.cls.m.find_and_modify(! {'_id': sem_id},! update={'$pullAll': {'queued': wake_msg_ids}},! new=True)! for msgid in wake_msg_ids:! make_dispatchable(msgid)! sem = updated Actually release Awaken queued message(s) Some magic (covered later)
99. @rick446 @synappio Message States ready acquire queued busy
100. @rick446 @synappio Message States ready acquire queued busy • Reserve the message
101. @rick446 @synappio Message States ready acquire queued busy • Reserve the message • Acquire resources
102. @rick446 @synappio Message States ready acquire queued busy • Reserve the message • Acquire resources • Process the message
103. @rick446 @synappio Message States ready acquire queued busy • Reserve the message • Acquire resources • Process the message • Release resources
104. @rick446 @synappio Reserve a Message msg = db.message.find_and_modify(! {'s.status': 'ready'},! sort=[('s.sub_status', -1), ('s.pri', -1), ('s.ts', 1)],! update={'$set': {'s.w': worker, 's.status': 'acquire'}},! new=True) message.s == {! pri: 10,! semaphores: ['foo'],! status: 'ready',! sub_status: 0,! w: '----------',! ...} message.s == {! pri: 10,! semaphores: ['foo'],! status: 'acquire! sub_status: 0,! w: worker,! ...}
105. @rick446 @synappio Reserve a Message msg = db.message.find_and_modify(! {'s.status': 'ready'},! sort=[('s.sub_status', -1), ('s.pri', -1), ('s.ts', 1)],! update={'$set': {'s.w': worker, 's.status': 'acquire'}},! new=True) message.s == {! pri: 10,! semaphores: ['foo'],! status: 'ready',! sub_status: 0,! w: '----------',! ...} Required semaphores message.s == {! pri: 10,! semaphores: ['foo'],! status: 'acquire! sub_status: 0,! w: worker,! ...}
106. @rick446 @synappio Reserve a Message msg = db.message.find_and_modify(! {'s.status': 'ready'},! sort=[('s.sub_status', -1), ('s.pri', -1), ('s.ts', 1)],! update={'$set': {'s.w': worker, 's.status': 'acquire'}},! new=True) message.s == {! pri: 10,! semaphores: ['foo'],! status: 'ready',! sub_status: 0,! w: '----------',! ...} Required semaphores # semaphores acquired message.s == {! pri: 10,! semaphores: ['foo'],! status: 'acquire! sub_status: 0,! w: worker,! ...}
107. @rick446 @synappio Reserve a Message msg = db.message.find_and_modify(! {'s.status': 'ready'},! sort=[('s.sub_status', -1), ('s.pri', -1), ('s.ts', 1)],! update={'$set': {'s.w': worker, 's.status': 'acquire'}},! new=True) message.s == {! pri: 10,! semaphores: ['foo'],! status: 'ready',! sub_status: 0,! w: '----------',! ...} Required semaphores # semaphores acquired message.s == {! pri: 10,! semaphores: ['foo'],! status: 'acquire! sub_status: 0,! w: worker,! ...} Prefer partially-acquired messages
108. @rick446 @synappio Acquire Resources def acquire_resources(msg):! for i, sem_id in enumerate(msg['s']['semaphores']):! if i < msg['sub_status']: # already acquired! continue! sem = db.semaphore.find_one({'_id': 'sem_id'})! if try_acquire_resource(sem_id, msg['_id'], sem['value']):! db.message.update(! {'_id': msg['_id']}, {'$set': {'s.sub_status': i}})! else:! return False! db.message.update(! {'_id': msg['_id']}, {'$set': {'s.status': 'busy'}})! return True
109. @rick446 @synappio Acquire Resources def acquire_resources(msg):! for i, sem_id in enumerate(msg['s']['semaphores']):! if i < msg['sub_status']: # already acquired! continue! sem = db.semaphore.find_one({'_id': 'sem_id'})! if try_acquire_resource(sem_id, msg['_id'], sem['value']):! db.message.update(! {'_id': msg['_id']}, {'$set': {'s.sub_status': i}})! else:! return False! db.message.update(! {'_id': msg['_id']}, {'$set': {'s.status': 'busy'}})! return True Save forward progress
110. @rick446 @synappio Acquire Resources def acquire_resources(msg):! for i, sem_id in enumerate(msg['s']['semaphores']):! if i < msg['sub_status']: # already acquired! continue! sem = db.semaphore.find_one({'_id': 'sem_id'})! if try_acquire_resource(sem_id, msg['_id'], sem['value']):! db.message.update(! {'_id': msg['_id']}, {'$set': {'s.sub_status': i}})! else:! return False! db.message.update(! {'_id': msg['_id']}, {'$set': {'s.status': 'busy'}})! return True Save forward progress Failure to acquire (already queued)
111. @rick446 @synappio Acquire Resources def acquire_resources(msg):! for i, sem_id in enumerate(msg['s']['semaphores']):! if i < msg['sub_status']: # already acquired! continue! sem = db.semaphore.find_one({'_id': 'sem_id'})! if try_acquire_resource(sem_id, msg['_id'], sem['value']):! db.message.update(! {'_id': msg['_id']}, {'$set': {'s.sub_status': i}})! else:! return False! db.message.update(! {'_id': msg['_id']}, {'$set': {'s.status': 'busy'}})! return True Save forward progress Failure to acquire (already queued) Resources acquired, message ready to be processed
112. @rick446 @synappio Acquire Resources def try_acquire_resource(sem_id, msg_id, sem_size):! '''Version 1 (race condition)'''! if reserve(sem_id, msg_id, sem_size):! return True! else:! db.message.update(! {'_id': msg_id},! {'$set': {'s.status': 'queued'}})! return False
113. @rick446 @synappio Acquire Resources def try_acquire_resource(sem_id, msg_id, sem_size):! '''Version 1 (race condition)'''! if reserve(sem_id, msg_id, sem_size):! return True! else:! db.message.update(! {'_id': msg_id},! {'$set': {'s.status': 'queued'}})! return False Here be dragons!
114. @rick446 @synappio Release Resources (v1) “magic” def make_dispatchable(msg_id):! '''Version 1 (race condition)'''! db.message.update(! {'_id': msg_id, 's.status': 'queued'},! {'$set': {'s.status': 'ready'}})
115. @rick446 @synappio Release Resources (v1) “magic” def make_dispatchable(msg_id):! '''Version 1 (race condition)'''! db.message.update(! {'_id': msg_id, 's.status': 'queued'},! {'$set': {'s.status': 'ready'}}) But what if s.status == ‘acquire’?
116. @rick446 @synappio Release Resources (v1) “magic” def make_dispatchable(msg_id):! '''Version 1 (race condition)'''! db.message.update(! {'_id': msg_id, 's.status': 'queued'},! {'$set': {'s.status': 'ready'}}) But what if s.status == ‘acquire’?
117. @rick446 @synappio Release Resources (v1) “magic” def make_dispatchable(msg_id):! '''Version 1 (race condition)'''! db.message.update(! {'_id': msg_id, 's.status': 'queued'},! {'$set': {'s.status': 'ready'}}) But what if s.status == ‘acquire’? That’s the dragon.
118. @rick446 @synappio Release Resources (v2) def make_dispatchable(msg_id):! res = db.message.update(! {'_id': msg_id, 's.status': 'acquire'},! {'$set': {'s.event': True}})! if not res['updatedExisting']:! db.message.update(! {'_id': msg_id, 's.status': 'queued'},! {'$set': {'s.status': 'ready'}})
119. @rick446 @synappio Release Resources (v2) def make_dispatchable(msg_id):! res = db.message.update(! {'_id': msg_id, 's.status': 'acquire'},! {'$set': {'s.event': True}})! if not res['updatedExisting']:! db.message.update(! {'_id': msg_id, 's.status': 'queued'},! {'$set': {'s.status': 'ready'}}) Hey, something happened!
120. @rick446 @synappio Acquire Resources (v2) def try_acquire_resource(sem_id, msg_id, sem_size):! '''Version 2'''! while True:! db.message.update(! {'_id': msg_id}, {'$set': {'event': False}})! if reserve(sem_id, msg_id, sem_size):! return True! else:! res = db.message.update(! {'_id': msg_id, 's.event': False},! {'$set': {'s.status': 'queued'}})! if not res['updatedExisting']:! # Someone released this message; try again! continue! return False
121. @rick446 @synappio Acquire Resources (v2) def try_acquire_resource(sem_id, msg_id, sem_size):! '''Version 2'''! while True:! db.message.update(! {'_id': msg_id}, {'$set': {'event': False}})! if reserve(sem_id, msg_id, sem_size):! return True! else:! res = db.message.update(! {'_id': msg_id, 's.event': False},! {'$set': {'s.status': 'queued'}})! if not res['updatedExisting']:! # Someone released this message; try again! continue! return False Nothing’s happened yet!
122. @rick446 @synappio Acquire Resources (v2) def try_acquire_resource(sem_id, msg_id, sem_size):! '''Version 2'''! while True:! db.message.update(! {'_id': msg_id}, {'$set': {'event': False}})! if reserve(sem_id, msg_id, sem_size):! return True! else:! res = db.message.update(! {'_id': msg_id, 's.event': False},! {'$set': {'s.status': 'queued'}})! if not res['updatedExisting']:! # Someone released this message; try again! continue! return False Nothing’s happened yet! Check if something happened
123. @rick446 @synappio One More Race…. def release(sem_id, msg_id, sem_size):! sem = db.semaphore.find_and_modify(! {'_id': sem_id},! update={'$pull': {! 'active': msg_id, ! 'queued': msg_id}},! new=True)! ! while len(sem['active']) < sem_size and sem['queued']:! wake_msg_ids = sem['queued'][:sem_size]! updated = self.cls.m.find_and_modify(! {'_id': sem_id},! update={'$pullAll': {'queued': wake_msg_ids}},! new=True)! for msgid in wake_msg_ids:! make_dispatchable(msgid)! sem = updated
124. @rick446 @synappio One More Race…. def release(sem_id, msg_id, sem_size):! sem = db.semaphore.find_and_modify(! {'_id': sem_id},! update={'$pull': {! 'active': msg_id, ! 'queued': msg_id}},! new=True)! ! while len(sem['active']) < sem_size and sem['queued']:! wake_msg_ids = sem['queued'][:sem_size]! updated = self.cls.m.find_and_modify(! {'_id': sem_id},! update={'$pullAll': {'queued': wake_msg_ids}},! new=True)! for msgid in wake_msg_ids:! make_dispatchable(msgid)! sem = updated
125. @rick446 @synappio Compensate! def fixup_queued_messages():! for msg in db.message.find({'s.status': 'queued'}):! sem_id = msg['semaphores'][msg['s']['sub_status']]! sem = db.semaphore.find_one(! {'_id': sem_id, 'queued': msg['_id']})! if sem is None:! db.message.m.update(! {'_id': msg['_id'], ! 's.status': 'queued', ! 's.sub_status': msg['sub_status']},! {'$set': {'s.status': 'ready'}})
126. @rick446 @synappio Managing Latency
127. @rick446 @synappio Managing Latency • Reserving messages is expensive • Use Pub/Sub system instead • Publish to the channel whenever a message is ready to be handled • Each worker subscribes to the channel • Workers only ‘poll’ when they have a chance of getting work
128. Capped Collections Capped Collection • Fixed size • Fast inserts • “Tailable” cursors Tailable Cursor
129. Capped Collections Capped Collection • Fixed size • Fast inserts • “Tailable” cursors Tailable Cursor
130. Capped Collections Capped Collection • Fixed size • Fast inserts • “Tailable” cursors Tailable Cursor
131. Capped Collections Capped Collection • Fixed size • Fast inserts • “Tailable” cursors Tailable Cursor
132. Capped Collections Capped Collection • Fixed size • Fast inserts • “Tailable” cursors Tailable Cursor
133. Capped Collections Capped Collection • Fixed size • Fast inserts • “Tailable” cursors Tailable Cursor
134. @rick446 @synappio Getting a Tailable Cursor def get_cursor(collection, topic_re, await_data=True):! options = { 'tailable': True }! if await_data:! options['await_data'] = True! cur = collection.find(! { 'k': topic_re },! **options)! cur = cur.hint([('$natural', 1)]) # ensure we don't use any indexes! return cur
135. @rick446 @synappio Getting a Tailable Cursor def get_cursor(collection, topic_re, await_data=True):! options = { 'tailable': True }! if await_data:! options['await_data'] = True! cur = collection.find(! { 'k': topic_re },! **options)! cur = cur.hint([('$natural', 1)]) # ensure we don't use any indexes! return cur Make cursor tailable
136. @rick446 @synappio Getting a Tailable Cursor def get_cursor(collection, topic_re, await_data=True):! options = { 'tailable': True }! if await_data:! options['await_data'] = True! cur = collection.find(! { 'k': topic_re },! **options)! cur = cur.hint([('$natural', 1)]) # ensure we don't use any indexes! return cur Holds open cursor for a while Make cursor tailable
137. @rick446 @synappio Getting a Tailable Cursor def get_cursor(collection, topic_re, await_data=True):! options = { 'tailable': True }! if await_data:! options['await_data'] = True! cur = collection.find(! { 'k': topic_re },! **options)! cur = cur.hint([('$natural', 1)]) # ensure we don't use any indexes! return cur Holds open cursor for a while Make cursor tailable Don’t use indexes
138. @rick446 @synappio Getting a Tailable Cursor def get_cursor(collection, topic_re, await_data=True):! options = { 'tailable': True }! if await_data:! options['await_data'] = True! cur = collection.find(! { 'k': topic_re },! **options)! cur = cur.hint([('$natural', 1)]) # ensure we don't use any indexes! return cur import re, time! while True:! cur = get_cursor(! db.capped_collection, ! re.compile('^foo'), ! await_data=True)! for msg in cur:! do_something(msg)! time.sleep(0.1) Holds open cursor for a while Make cursor tailable Don’t use indexes
139. @rick446 @synappio Getting a Tailable Cursor def get_cursor(collection, topic_re, await_data=True):! options = { 'tailable': True }! if await_data:! options['await_data'] = True! cur = collection.find(! { 'k': topic_re },! **options)! cur = cur.hint([('$natural', 1)]) # ensure we don't use any indexes! return cur import re, time! while True:! cur = get_cursor(! db.capped_collection, ! re.compile('^foo'), ! await_data=True)! for msg in cur:! do_something(msg)! time.sleep(0.1) Holds open cursor for a while Make cursor tailable Don’t use indexes Still some polling when no producer, so don’t spin too fast
140. @rick446 @synappio Building in retry... def get_cursor(collection, topic_re, last_id=-1, await_data=True):! options = { 'tailable': True }! spec = { ! 'id': { '$gt': last_id }, # only new messages! 'k': topic_re }! if await_data:! options['await_data'] = True! cur = collection.find(spec, **options)! cur = cur.hint([('$natural', 1)]) # ensure we don't use any indexes! return cur
141. @rick446 @synappio Building in retry... def get_cursor(collection, topic_re, last_id=-1, await_data=True):! options = { 'tailable': True }! spec = { ! 'id': { '$gt': last_id }, # only new messages! 'k': topic_re }! if await_data:! options['await_data'] = True! cur = collection.find(spec, **options)! cur = cur.hint([('$natural', 1)]) # ensure we don't use any indexes! return cur Integer autoincrement “id”
142. @rick446 @synappio Ludicrous Speed from pymongo.cursor import _QUERY_OPTIONS! ! def get_cursor(collection, topic_re, last_id=-1, await_data=True):! options = { 'tailable': True }! spec = { ! 'ts': { '$gt': last_id }, # only new messages! 'k': topic_re }! if await_data:! options['await_data'] = True! cur = collection.find(spec, **options)! cur = cur.hint([('$natural', 1)]) # ensure we don't use any indexes! if await:! cur = cur.add_option(_QUERY_OPTIONS['oplog_replay'])! return cur
143. @rick446 @synappio Ludicrous Speed from pymongo.cursor import _QUERY_OPTIONS! ! def get_cursor(collection, topic_re, last_id=-1, await_data=True):! options = { 'tailable': True }! spec = { ! 'ts': { '$gt': last_id }, # only new messages! 'k': topic_re }! if await_data:! options['await_data'] = True! cur = collection.find(spec, **options)! cur = cur.hint([('$natural', 1)]) # ensure we don't use any indexes! if await:! cur = cur.add_option(_QUERY_OPTIONS['oplog_replay'])! return cur id ==> ts
144. @rick446 @synappio Ludicrous Speed from pymongo.cursor import _QUERY_OPTIONS! ! def get_cursor(collection, topic_re, last_id=-1, await_data=True):! options = { 'tailable': True }! spec = { ! 'ts': { '$gt': last_id }, # only new messages! 'k': topic_re }! if await_data:! options['await_data'] = True! cur = collection.find(spec, **options)! cur = cur.hint([('$natural', 1)]) # ensure we don't use any indexes! if await:! cur = cur.add_option(_QUERY_OPTIONS['oplog_replay'])! return cur id ==> ts Co-opt the oplog_replay option
145. @rick446 @synappio The Oplog • Capped collection that records all operations for replication • Includes a ‘ts’ field suitable for oplog_replay • Does not require a separate publish operation (all changes are automatically “published”)
146. @rick446 @synappio Using the Oplog def oplog_await(oplog, spec):! '''Await the very next message on the oplog satisfying the spec'''! last = oplog.find_one(spec, sort=[('$natural', -1)])! if last is None:! return # Can't await unless there is an existing message satisfying spec! await_spec = dict(spec)! last_ts = last['ts']! await_spec['ts'] = {'$gt': bson.Timestamp(last_ts.time, last_ts.inc - 1)}! curs = oplog.find(await_spec, tailable=True, await_data=True)! curs = curs.hint([('$natural', 1)])! curs = curs.add_option(_QUERY_OPTIONS['oplog_replay'])! curs.next() # should always find 1 element! try:! return curs.next()! except StopIteration:! return None
147. @rick446 @synappio Using the Oplog def oplog_await(oplog, spec):! '''Await the very next message on the oplog satisfying the spec'''! last = oplog.find_one(spec, sort=[('$natural', -1)])! if last is None:! return # Can't await unless there is an existing message satisfying spec! await_spec = dict(spec)! last_ts = last['ts']! await_spec['ts'] = {'$gt': bson.Timestamp(last_ts.time, last_ts.inc - 1)}! curs = oplog.find(await_spec, tailable=True, await_data=True)! curs = curs.hint([('$natural', 1)])! curs = curs.add_option(_QUERY_OPTIONS['oplog_replay'])! curs.next() # should always find 1 element! try:! return curs.next()! except StopIteration:! return None most recent oplog entry
148. @rick446 @synappio Using the Oplog def oplog_await(oplog, spec):! '''Await the very next message on the oplog satisfying the spec'''! last = oplog.find_one(spec, sort=[('$natural', -1)])! if last is None:! return # Can't await unless there is an existing message satisfying spec! await_spec = dict(spec)! last_ts = last['ts']! await_spec['ts'] = {'$gt': bson.Timestamp(last_ts.time, last_ts.inc - 1)}! curs = oplog.find(await_spec, tailable=True, await_data=True)! curs = curs.hint([('$natural', 1)])! curs = curs.add_option(_QUERY_OPTIONS['oplog_replay'])! curs.next() # should always find 1 element! try:! return curs.next()! except StopIteration:! return None most recent oplog entry finds most recent plus following entries
149. @rick446 @synappio Using the Oplog def oplog_await(oplog, spec):! '''Await the very next message on the oplog satisfying the spec'''! last = oplog.find_one(spec, sort=[('$natural', -1)])! if last is None:! return # Can't await unless there is an existing message satisfying spec! await_spec = dict(spec)! last_ts = last['ts']! await_spec['ts'] = {'$gt': bson.Timestamp(last_ts.time, last_ts.inc - 1)}! curs = oplog.find(await_spec, tailable=True, await_data=True)! curs = curs.hint([('$natural', 1)])! curs = curs.add_option(_QUERY_OPTIONS['oplog_replay'])! curs.next() # should always find 1 element! try:! return curs.next()! except StopIteration:! return None most recent oplog entry finds most recent plus following entries skip most recent
150. @rick446 @synappio Using the Oplog def oplog_await(oplog, spec):! '''Await the very next message on the oplog satisfying the spec'''! last = oplog.find_one(spec, sort=[('$natural', -1)])! if last is None:! return # Can't await unless there is an existing message satisfying spec! await_spec = dict(spec)! last_ts = last['ts']! await_spec['ts'] = {'$gt': bson.Timestamp(last_ts.time, last_ts.inc - 1)}! curs = oplog.find(await_spec, tailable=True, await_data=True)! curs = curs.hint([('$natural', 1)])! curs = curs.add_option(_QUERY_OPTIONS['oplog_replay'])! curs.next() # should always find 1 element! try:! return curs.next()! except StopIteration:! return None most recent oplog entry finds most recent plus following entries skip most recent return on anything new
151. @rick446 @synappio What We’ve Learned
152. @rick446 @synappio What We’ve Learned How to…
153. @rick446 @synappio What We’ve Learned How to… Build a task queue in MongoDB
154. @rick446 @synappio What We’ve Learned How to… Build a task queue in MongoDB
155. @rick446 @synappio What We’ve Learned How to… Build a task queue in MongoDB Bring consistency to distributed systems (without transactions)
156. @rick446 @synappio What We’ve Learned How to… Build a task queue in MongoDB Bring consistency to distributed systems (without transactions)
157. @rick446 @synappio What We’ve Learned How to… Build a task queue in MongoDB Bring consistency to distributed systems (without transactions) Build low-latency reactive systems
158. @rick446 @synappio Tips
159. @rick446 @synappio Tips • findAndModify is ideal for queues
160. @rick446 @synappio Tips • findAndModify is ideal for queues
161. @rick446 @synappio Tips • findAndModify is ideal for queues • Atomic update + compensation brings consistency to your distributed system
162. @rick446 @synappio Tips • findAndModify is ideal for queues • Atomic update + compensation brings consistency to your distributed system
163. @rick446 @synappio Tips • findAndModify is ideal for queues • Atomic update + compensation brings consistency to your distributed system • Use the oplog to build reactive, low-latency systems
164. Questions? Rick Copeland rick@synapp.io @rick446