GeezerVisonThe World Wide Web, restored to 800 pixels of glory. |
| Home|About|Gaming cabinet|Archives|How it works|Starter sites |
GeezerVison is showing github.com — Concurrency Concurrency
poolitzer edited this page
PTB is built on top of Python's Note:
By default, incoming updates and handler callbacks are processed sequentially, i.e. one after the other. So, if one callback function takes some time to execute, all other updates have to wait for it. Example:
You're running the Echobot and two users (User A and User B) send a message to the bot at the same time.
Maybe User A was a bit quicker, so their request arrives first, in the form of an If you have handlers in multiple groups, it gets a tiny bit more complicated.
The following pseudocode explains how async def process_update(self, update): # self is the `Application` instance for group_number, handlers in self.handlers.items(): # `handlers` is the list of handlers in group `group_number` for handler in handlers: if handler.check_update(update): # Here we `await`, i.e. we only continue after the callback is done! await handler.handle_update(update) break # at most one handler per group handles the update We want to reply to both User A and User B as fast as possible and while sending the reply to user User A we'd like to already get started on handling Update B. PTB comes with three built-in mechanisms that can help with that. Via the application.add_handler( MessageHandler(filters.TEXT & ~filters.COMMAND, echo, block=False) ) Instead, it will run the callback as Again, let's have a look at pseudocode: async def process_update(self, update): # self is the `Application` instance for group_number, handlers in self.handlers.items(): # `handlers` is the list of handlers in group `group_number` for handler in handlers: if handler.check_update(update): # Here we *don't* `await`, such that the loop immediately continues asyncio.create_task(handler.handle_update(update)) break # at most one handler per group handles the update This already helps for many use cases.
However, by using Instead of running single handlers in a non-blocking way, we can tell the Application.builder().token('TOKEN').concurrent_updates(True).build() Now the while not application.update_queue.empty(): update = await application.update_queue.get() asyncio.create_task(application.process_update(update)) This setting is independent of the You can further customize concurrent handling class MyUpdateProcessor(BaseUpdateProcessor): async def do_process_update(self, update, coroutine) -> None: # This method is called for every update if update.callback_query: await asyncio.sleep(5) await coroutine async def initialize(self) -> None: pass async def shutdown(self) -> None: pass Application.builder().token('TOKEN').concurrent_updates(MyUpdateProcessor(10)).build() The above code processes every while not application.update_queue.empty(): update = await application.update_queue.get() coroutine = application.process_update(update) asyncio.create_task(my_update_processor.do_process_update(update, coroutine)) This is just an example of how to use the Note: The number of concurrently processed updates is limited (the limit defaults to 4096 updates at a time). This is a simple measure to avoid e.g. DDOS attacks
This wrapper gives you fine-grained control about how you use concurrency in PTB. The next section gives you an idea about why that is helpful. Even though Probably the biggest cause of issues of concurrency are shared states, and those issues are hard to fix. So instead of showing you how to fix them, we'll show you how to avoid them altogether. More about that later. A fair warning: In this section, we'll try to give you a simple talk (if that's possible) on a very complex topic. Many have written about it before, and we're certainly less qualified than most. As usual, we'll use an example to complement the text, and try to stay in the realm of what's important to you. An example that is often used to illustrate this is that of a bank.
Let's say you have been hired by a bank to write a Telegram bot to manage bank accounts. The bot has the command async def transaction(update, context): bot = context.bot chat_id = update.effective_user.id source_id, target_id, amount = parse_update(update) await bot.send_message(chat_id, 'Preparing...') bank.log(BEGINNING_TRANSACTION, amount, source_id, target_id) source = bank.read_account(source_id) target = bank.read_account(target_id) source.balance -= amount target.balance += amount await bot.send_message(chat_id, 'Transferring money...') bank.log(CALCULATED_TRANSACTION, amount, source_id, target_id) bank.write_account(source) await bot.send_message(chat_id, 'Source account updated...') await bot.send_message(chat_id, 'Target account updated...') bank.write_account(target) await bot.send_message(chat_id, 'Done!') bank.log(FINISHED_TRANSACTION, amount, source_id, target_id) application.add_handler(CommandHandler('transaction', transaction, block=False)) We skipped some of the implementation details, so here's a short explanation:
Sadly, Now the
In the end, Account A is at +$100 and Account B is at +$10. Of course, this won't happen very often. And that's what makes this bug so critical. It will probably be missed by your tests and end up in production, potentially causing a lot of financial damage. Note: This kind of bug is called a race condition and has been the source of many, many security vulnerabilities. It's also one of the reasons why banking software is not written by unpaid interns. To be fair, you probably don't write software for banks (if you do, you should already know about this), but this kind of bug can occur in much simpler situations.
While in this case, the shared state is the As promised in the first paragraph, let's discuss how to avoid such situations. That's not always as easy as it is in this case, but we're lucky:
There are two relatively simple steps you have to follow. First, identify those parts of the code that must run sequentially (the opposite of in parallel or concurrently). Usually, that is code that fits at least one of these criteria:
Make sure you have a good idea what shared state means Don't hesitate to do a quick Google search on it. Let's go through our bank example line by line and note which of the criteria it matches: async def transaction(update, context): bot = context.bot chat_id = update.effective_user.id # 3 source_id, target_id, amount = parse_update(update) # 3 await bot.send_message(chat_id, 'Preparing...') # None bank.log(BEGINNING_TRANSACTION, amount, source_id, target_id) # None source = bank.read_account(source_id) # 2, 3 target = bank.read_account(target_id) # 2, 3 source.balance -= amount # 3 target.balance += amount # 3 await bot.send_message(chat_id, 'Transferring money...') # None bank.log(CALCULATED_TRANSACTION, amount, source_id, target_id) # None bank.write_account(source) # 1 await bot.send_message(chat_id, 'Source account updated...') # None await bot.send_message(chat_id, 'Target account updated...') # None bank.write_account(target) # 1 await bot.send_message(chat_id, 'Done!') # None bank.log(FINISHED_TRANSACTION, amount, source_id, target_id) # None application.add_handler(CommandHandler('transaction', transaction, block=False)) Note:
One could argue that As you can see, there's a pretty obvious pattern here:
async def log_and_notify(action, amount, source_id, target_id, chat_id, message): bank.log(action, amount, source_id, target_id) await bot.send_message(chat_id, message) async def transaction(update, context): chat_id = update.message.chat_id # 3 source_id, target_id, amount = parse_update(update) # 3 context.application.create_task( log_and_notify( BEGINNING_TRANSACTION, amount, source_id, target_id, chat_id, 'Preparing...', ), update=update ) source = bank.read_account(source_id) # 2, 3 target = bank.read_account(target_id) # 2, 3 source.balance -= amount # 3 target.balance += amount # 3 context.application.create_task( log_and_notify( CALCULATED_TRANSACTION, amount, source_id, target_id, chat_id, 'Transferring money...' ), update=update ) bank.write_account(source) # 1 bank.write_account(target) # 1 context.application.create_task( log_and_notify( FINISHED_TRANSACTION, amount, source_id, target_id, chat_id, 'Done!', ), update=update ) application.add_handler(CommandHandler('transaction', transaction, block=True)) Note: You might have noticed that we moved Note: It's likely that By separating the strictly sequential code from the concurrent code, we made sure that no race conditions can occur.
The That's basically it for now. For the very end, here are a few helpful guidelines for writing concurrent code:
Wiki of
Clone this wiki locally |
| You are visitor 01402503 — GeezerVison does not store the pages you look up. Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required. |