PyPI 官网下载 | wrk-0.3.2.tar.gz

# Kale: Distributed task worker from Nextdoor  [](https://travis-ci.org/Nextdoor/ndkale) Kale is a python task worker library that supports priority queues on Amazon SQS. Check out our blog post - [Nextdoor Taskworker: Simple, Efficient & Scalable](https://engblog.nextdoor.com/ac4f7886957b) ## How does it work?  Like other distributed task queue system, publishers send task messages to queues and workers fetch messages from queues. For now, Kale supports only Amazon SQS for the queue. ### Publisher A publisher can be any python program that imports a Kale-based task class and invokes the publish function of this class. For example, if a task class looks like this: # tasks.py class MyTask: def run_task(self, arg1, arg2, *args, **kwargs): # Do something Then the publisher publishes a task to Amazon SQS, which normally takes 10s miliseconds to return: import tasks tasks.MyTask.publish(None, arg1, arg2) The publish() function is a static method of a task class. Other than the first parameter, which can usually be `None`, it has the same signiture as the `run_task()` method. A worker process, which may run on a different machine, will pick up the message and execute `run_task()` method of the task. While ndkale is usable out of the box, the first parameter in `publish(app_data, *args, *kwargs)` is designed for more complex situations where certain state may need to be passed outside the context of the actual task parameters. One example of this might be to pass the environment. The `app_data` must be pickleable so that in can be encoded and inserted into the SQS message. The default task object will be populated with an `app_data` attribute, but the default worker will not use it. You will need to extend the default Worker or Task class to take advantage of `app_data`. ### Worker  A worker process runs an infinite loop. For each iteration, it does the following things: 1. It runs a **queue selection algorithm** (select_queue) to decides which queue to fetch tasks from; 2. It fetches a batch of tasks from a queue (get_messages); 3. It runs tasks one by one in the same batch (run_task); 4. Finish up. 1. If a task succeeded, it'll be deleted from the queue; 2. If a task runs too long (exceeding **time_limit** that is a per task property for task SLA) or it fails, it'll be put back to the queue and other workers will pick it up in the future (if retry is allowed); 3. If a batch of tasks runs too long, exceeding **visibility timeout** that is a per queue property for task batch SLA, then unfinished tasks will be put back to the queue and other workers will pick them up in the future. 5. It exits this iteration and enters next iteration and repeat the above steps. #### Queue Selection Algorithm Code: kale/queue_selector.py A good queue selection algorithm has these requirements: 1. higher priority queues should have more chances to be selected than lower priority queues; 2. it should not starve low priority queues; 3. it should not send too many requests to SQS while retrieving no task, which is waste of compute resource and Amazon charges us by the number of requests. 4. it should not wait on empty queues for too long, avoiding waste of compute resources. We experimented and benchmarked quite a few queue selection algorithms. We end up using an improved version of lottery algorithm, **ReducedLottery**, which fulfill the above requirements. ReducedLottery works like this: Initialize the lottery pool with all queues while lottery pool is not empty: Run lottery based on queue priority to get a queue who wins the jackpot Short poll SQS to see if the selected queue is empty if the selected queue is not empty: return queue else: Remove this queue from the lottery pool Reset the lottery pool with all queues Return whatever queue who wins the jackpot The beauty of **ReducedLottery**: * It prefers higher priority queues, as higher priority queues get more lottery tickets and have higher chances to win the jackpot. Thus, requirement 1 is fulfilled. * It uses randomness to avoid starvation. Lower priority queues still have chance to win the jackpot. Thus, requirement 2 is fulfilled. * If the selected queue is empty, SQS will automatically let task worker long poll on the queue, avoiding sending too many requests (short polls). Thus, requirement 3 is fulfilled. * It excludes known empty queues from the lottery pool. Only when all queues are empty can it returns an empty queue. So, it's unlikely to long poll on an empty queue. Thus, requirement 4 is fulfilled. #### Settings There are two types of settings, worker config and queue config. ##### Worker config Settings are specified in settings modules, including AWS confidentials, queues config, queue selection algorithm, ... Settings modules are loaded in such order: * kale.default_settings * the module specified via KALE\_SETTINGS\_MODULE environment variable Here's an example import os AWS_REGION = 'us-west-2' # # Production settings # (use this for prod to talk to Amazon SQS) # MESSAGE_QUEUE_USE_PROXY = False # AWS_ACCESS_KEY_ID = 'AWS KEY ID' # AWS_SECRET_ACCESS_KEY = ''AWS SECRET KEY # # Development settings # (use this for dev to talk to ElasticMQ, which is SQS emulator) # Using elasticmq to emulate SQS locally MESSAGE_QUEUE_USE_PROXY = True MESSAGE_QUEUE_PROXY_PORT = 9324 MESSAGE_QUEUE_PROXY_HOST = os.getenv('MESSAGE_QUEUE_PROXY_HOST', '0.0.0.0') AWS_ACCESS_KEY_ID = 'x' AWS_SECRET_ACCESS_KEY = 'x' QUEUE_CONFIG = 'taskworker/queue_config.yaml' # SQS limits per message size, bytes # It can be set anywhere from 1024 bytes (1KB), up to 262144 bytes (256KB). # See http://aws.amazon.com/sqs/faqs/ SQS_TASK_SIZE_LIMIT = 256000 QUEUE_SELECTOR = 'kale.queue_selector.ReducedLottery' Settings in the later modules overwrite those in the early-loaded modules. ##### Queue config All queues and their properties are in a queues config yaml file whose path is specified in the above settings modules. Here's an example # task SLA: 60/10 = 6 seconds high_priority: name: high_priority priority: 100 batch_size: 10 visibility_timeout_sec: 60 long_poll_time_sec: 1 num_iterations: 10 # task SLA: 60 / 10 = 6 seconds default: name: default priority: 40 batch_size: 10 visibility_timeout_sec: 60 long_poll_time_sec: 1 num_iterations: 5 # task SLA: 60 / 10 = 6 seconds low_priority: name: low_priority priority: 5 batch_size: 10 visibility_timeout_sec: 60 long_poll_time_sec: 5 num_iterations: 5 ## How to implement a distributed task worker system using Kale ### Install kale From source code python setup.py install Using pip (from github repo) # # Put this in requirements.txt, then run # pip install -r requirements.txt # # If you want the latest build git+https://github.com/Nextdoor/ndkale.git#egg=ndkale # Or put this if you want a specific commit git+https://github.com/Nextdoor/ndkale.git@67f873ed7b0a8131cc8d72453d749ffb389d69