Reflected SQLAlchemy metadata in celery tasks?

For better testability and other reasons, it is good to have SQLAlchemy database sessions configuration non-global as described very well in the following question:

how to setup sqlalchemy session in celery tasks with no global variable (and also discussed in https://github.com/celery/celery/issues/3561 )

Now, the question is, how to handle metadata elegantly? If my understanding is correct, metadata can be had once, eg:

engine = create_engine(DB_URL, encoding='utf-8', pool_recycle=3600,                        pool_size=10) # db_session = get_session()  # this is old global session meta = MetaData() meta.reflect(bind=engine) 

Reflecting on each task execution is not good for performance reason, metadata is more or less stable and thread-safe structure (if we only read it).

However, metadata sometimes changes (celery is not the "owner" of the db schema), causing errors in workers.

What could be an elegant way to deal with meta in a testable way, plus still be able to react to underlying db changes? (alembic in use, if it is relevant).

I was thinking of using alembic version change as a signal to re-reflect, but not quite sure how to make it work nicely in celery. For instance, if more than one worker will at once sense a change, the global meta may be treated in a non-thread safety way.

If it matters, celery use in the case is standalone, no web framework modules/apps/whatever present in the celery app. The problem is also simplified as only SQLAlchemy Core is in use, not object mapper.

0 Answers

Read More

Using context managers for recovering from celery's SoftTimeLimitExceeded

I am trying to set a maximum run time for my celery jobs.

I am currently recovering from exceptions with a context manager. I ended up with code very similar to this snippet:

from celery.exceptions import SoftTimeLimitExceeded  class Manager:      def __enter__(self):         return self      def __exit__(self, error_type, error, tb):         if error_type == SoftTimeLimitExceeded:             logger.info('job killed.')             # swallow the exception             return True   @task def do_foo():     with Manager():         run_task1()         run_task2()         run_task3() 

What I expected:

If do_foo times out in run_task1, the logger logs, the SoftTimeLimitExceeded exception is swallowed, the body of the manager is skipped, the job ends without running run_task2 and run_task3.

What I observe: do_foo times out in run_task1, SoftTimeLimitExceeded is raised, the logger logs, the SoftTimeLimitExceeded exception is swallowed but run_task2 and run_task3 are running nevertheless.

I am looking for an answer to following two questions:

1. Why is run_task2 still executed when SoftTimeLimitExceeded is raised in run_task1 in this setting?

2. Is there an easy way to transform my code so that it can performs as expected?

1 Answers

Answers 1

Cleaning up the code

This code is pretty good; there's not much cleaning up to do.

• You shouldn't return self from __enter__ if the context manager isn't designed to be used with the as keyword.
• is should be used when checking classes, since they are singletons...
• but you should prefer issubclass to properly emulate exception handling.

Implementing these changes gives:

from celery.exceptions import SoftTimeLimitExceeded  class Manager:     def __enter__(self):         pass      def __exit__(self, error_type, error, tb):         if issubclass(error_type, SoftTimeLimitExceeded):             logger.info('job killed.')             # swallow the exception             return True  @task def do_foo():     with Manager():         run_task1()         run_task2()         run_task3() 

Debugging

I created a mock environment for debugging:

class SoftTimeLimitExceeded(Exception):     pass  class Logger:     info = print logger = Logger() del Logger  def task(f):     return f  def run_task1():     print("running task 1")     raise SoftTimeLimitExceeded  def run_task2():     print("running task 2")  def run_task_3():     print("running task 3") 

Executing this and then your program gives:

>>> do_foo() running task 1 job killed. 

This is the expected behaviour.

Hypotheses

I can think of two possibilities:

1. Something in the chain, probably run_task1, is asynchronous.
2. celery is doing something weird.

I'll run with the second hypothesis because I can't test the former.

I've been bitten by the obscure behaviour of a combination between context managers, exceptions and coroutines before, so I know what sorts of problems it causes. This seems like one of them, but I'll have to look at celery's code before I can go any further.

Edit: I can't make head nor tail of celery's code, and searching hasn't turned up the code that raises SoftTimeLimitExceeded to allow me to trace it backwards. I'll pass it on to somebody more experienced with celery to see if they can work out how it works.

Read More

celery shutdown worker after particular task

I'm using celery (solo pool with concurrency=1) and I want to be able to shut down the worker after a particular task has run. A caveat is that I want to avoid any possibility of the worker picking up any further tasks after that one.

Here's my attempt in the outline:

from __future__ import absolute_import, unicode_literals from celery import Celery from celery.exceptions import WorkerShutdown from celery.signals import task_postrun  app = Celery() app.config_from_object('celeryconfig')  @app.task def add(x, y):     return x + y  @task_postrun.connect(sender=add) def shutdown(*args, **kwargs):     raise WorkerShutdown() 

However, when I run the worker

celery -A celeryapp  worker --concurrency=1 --pool=solo 

and run the task

add.delay(1,4) 

I get the following:

 -------------- celery@sam-APOLLO-2000 v4.0.2 (latentcall) ---- **** -----  --- * ***  * -- Linux-4.4.0-116-generic-x86_64-with-Ubuntu-16.04-xenial 2018-03-18 14:08:37 -- * - **** ---  - ** ---------- [config] - ** ---------- .> app:         __main__:0x7f596896ce90 - ** ---------- .> transport:   redis://localhost:6379/0 - ** ---------- .> results:     redis://localhost/ - *** --- * --- .> concurrency: 4 (solo) -- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker) --- ***** -----   -------------- [queues]                 .> celery           exchange=celery(direct) key=celery   [2018-03-18 14:08:39,892: WARNING/MainProcess] Restoring 1 unacknowledged message(s) 

The task is re-queued and will be run again on another worker, leading to a loop.

This also happens when I move the WorkerShutdown exception within the task itself.

@app.task def add(x, y):     print(x + y)     raise WorkerShutdown() 

Is there a way I can shut down the worker after a particular task, while avoiding this unfortunate side-effect?

2 Answers

Answers 1

The recommended process for shutting down a worker is to send the TERM signal. This will cause a celery worker to shutdown after completing any currently running tasks. If you send a QUIT signal to the worker's main process, the worker will shutdown immediately.

The celery docs, however, usually discuss this in terms of managing celery from a command line or via systemd/initd, but celery additionally provides a remote worker control API via celery.app.control.
You can revoke a task to prevent workers from executing the task. This should prevent the loop you are experiencing. Further, control supports shutdown of a worker in this manner as well.

So I imagine the following will get you the behavior you desire.

@app.task(bind=True) def shutdown(self):     app.control.revoke(self.id) # prevent this task from being executed again     app.control.shutdown() # send shutdown signal to all workers 

Since it's not currently possible to ack the task from within the task, then continue executing said task, this method of using revoke circumvents this problem so that, even if the task is queued again, the new worker will simply ignore it.

Alternatively, the following would also prevent a redelivered task from being executed a second time...

@app.task(bind=True) def some_task(self):     if self.request.delivery_info['redelivered']:         raise Ignore() # ignore if this task was redelivered     print('This should only execute on first receipt of task') 

Also worth noting AsyncResult also has a revoke method that calls self.app.control.revoke for you.

Answers 2

If you shutdown the worker, after the task has completed, it won't re-queue again.

@task_postrun.connect(sender=add) def shutdown(*args, **kwargs):     app.control.broadcast('shutdown') 

This will gracefully shutdown the worker after tasks is completed.

[2018-04-01 18:44:14,627: INFO/MainProcess] Connected to redis://localhost:6379/0 [2018-04-01 18:44:14,656: INFO/MainProcess] mingle: searching for neighbors [2018-04-01 18:44:15,719: INFO/MainProcess] mingle: all alone [2018-04-01 18:44:15,742: INFO/MainProcess] celery@foo ready. [2018-04-01 18:46:28,572: INFO/MainProcess] Received task: celery_worker_stop.add[ac8a65ff-5aad-41a6-a2d6-a659d021fb9b] [2018-04-01 18:46:28,585: INFO/ForkPoolWorker-4] Task celery_worker_stop.add[ac8a65ff-5aad-41a6-a2d6-a659d021fb9b] succeeded in 0.005628278013318777s: 3    [2018-04-01 18:46:28,665: WARNING/MainProcess] Got shutdown from remote 

Note: broadcast will shutdown all workers. If you want to shutdonw a specific worker, start worker with a name

celery -A celeryapp  worker -n self_killing --concurrency=1 --pool=solo 

Now you can shutdown this with destination parameter.

app.control.broadcast('shutdown', destination=['celery@self_killing']) 

Read More

limited number of user-initiated background processes

I need to allow users to submit requests for very, very large jobs. We are talking 100 gigabytes of memory and 20 hours of computing time. This costs our company a lot of money, so it was stipulated that only 2 jobs could be running at any time, and requests for new jobs when 2 are already running would be rejected (and the user notified that the server is busy).

My current solution uses an Executor from concurrent.futures, and requires setting the Apache server to run only one process, reducing responsiveness (current user count is very low, so it's okay for now).

If possible I would like to use Celery for this, but I did not see in the documentation any way to accomplish this particular setting.

How can I run up to a limited number of jobs in the background in a Django application, and notify users when jobs are rejected because the server is busy?

3 Answers

Answers 1

I have two solutions for this particular case, one an out of the box solution by celery, and another one that you implement yourself.

1. You can do something like this with celery workers. In particular, you only create two worker processes with concurrency=1 (or well, one with concurrency=2, but that's gonna be threads, not different processes), this way, only two jobs can be done asynchronously. Now you need a way to raise exceptions if both jobs are occupied, then you use inspect, to count the number of active tasks and throw exceptions if required. For implementation, you can checkout this SO post.

You might also be interested in rate limits.

2. You can do it all yourself, using a locking solution of choice. In particular, a nice implementation that makes sure only two processes are running with redis (and redis-py) is as simple as the following. (Considering you know redis, since you know celery)

   from redis import StrictRedis  redis = StrictRedis('localhost', '6379') locks = ['compute:lock1', 'compute:lock2'] for key in locks:     lock = redis.lock(key, blocking_timeout=5)     acquired = lock.acquire()     if acquired:         do_huge_computation()         lock.release()     else:         raise SystemLimitsReached("Already at max capacity !") 

This way you make sure only two running processes can exist in the system. A third processes will block in the line lock = redis.lock(key) for blocking_timeout seconds, if the locking was successful, acquired would be True, else it's False and you'd tell your user to wait !

I had the same requirement sometime in the past and what I ended up coding was something like the solution above. In particular

1. This has the least amount of race conditions possible
2. It's easy to read
3. Doesn't depend on a sysadmin, suddenly doubling the concurrency of workers under load and blowing up the whole system.
4. You can also implement the limit per user, meaning each user can have 2 simultaneous running jobs, by only changing the lock keys from compute:lock1 to compute:userId:lock1 and lock2 accordingly. You can't do this one with vanila celery.

Answers 2

First of all you need to limit concurrency on your worker (docs):

celery -A proj worker --loglevel=INFO --concurrency=2 -n <worker_name> 

This will help to make sure that you do not have more than 2 active tasks even if you will have errors in the code.

Now you have 2 ways to implement task number validation:

1. You can use inspect to get number of active and scheduled tasks:

    from celery import current_app   def start_job():       inspect = current_app.control.inspect()       active_tasks = inspect.active() or {}       scheduled_tasks = inspect.scheduled() or {}       worker_key = 'celery@%s' % <worker_name>       worker_tasks = active_tasks.get(worker_key, []) + scheduled_tasks.get(worker_key, [])       if len(worker_tasks) >= 2:           raise MyCustomException('It is impossible to start more than 2 tasks.')        else:           my_task.delay() 

2. You can store number of currently executing tasks in DB and validate task execution based on it.

Second approach could be better if you want to scale your functionality - introduce premium users or do not allow to execute 2 requests by one user.

Answers 3

First

You need the first part of SpiXel's solution. According to him, "you only create two worker processes with concurrency=1".

Second

Set the time out for the task waiting in the queue, which is set CELERY_EVENT_QUEUE_TTL and the queue length limit according to how to limit number of tasks in queue and stop feeding when full?.

Therefore, when the two work running jobs, and the task in the queue waiting like 10 sec or any period time you like, the task will be time out. Or if the queue has been fulfilled, new arrival tasks will be dropped out.

Third

you need extra things to deal with notifying "users when jobs are rejected because the server is busy".

Dead Letter Exchanges is what you need. Every time a task is failed because of the queue length limit or message timeout. "Messages will be dropped or dead-lettered from the front of the queue to make room for new messages once the limit is reached."

You can set "x-dead-letter-exchange" to route to another queue, once this queue receive the dead lettered message, you can send a notification message to users.

Read More

celery, flask sqlalchemy: DatabaseError: (DatabaseError) SSL error: decryption failed or bad record mac

Hi I have a setup where I'm using Celery Flask SqlAlchemy and I am intermittently getting this error:

 (psycopg2.DatabaseError) SSL error: decryption failed or bad record mac 

I followed this post:

Celery + SQLAlchemy : DatabaseError: (DatabaseError) SSL error: decryption failed or bad record mac

and also a few more and added a prerun and postrun methods:

@task_postrun.connect def close_session(*args, **kwargs):     # Flask SQLAlchemy will automatically create new sessions for you from      # a scoped session factory, given that we are maintaining the same app     # context, this ensures tasks have a fresh session (e.g. session errors      # won't propagate across tasks)     d.session.remove()  @task_prerun.connect def on_task_init(*args, **kwargs):     d.engine.dispose() 

But I'm still seeing this error. Anyone solved this?

Note that I'm running this on AWS (with two servers accessing same database). Database itself is hosted on it's own server (not RDS). I believe the total celery background tasks running are 6 (2+4). Flask frontend is running using gunicorn.

0 Answers

Read More

Networkx as a task queue?

I have a directed acyclic graph in networkx. Each node represents a task and a nodes' predecessors are task dependencies (a given task cannot execute until its' dependencies have executed).

I'd like to 'execute' the graph in an asynchronous task queue, similar to what celery offers (so that I can poll jobs for their status, retrieve results etc). Celery doesnt offer the ability to create DAG's (as far as I know) and having the ability to move on to a task as soon as all dependencies are complete would be crucial (a DAG may have multiple paths and even if one task is slow/blocking, it may be possible to move on to other tasks etc).

Are there any simple examples as to how I could achieve this, or perhaps even integrate networkx with celery?

1 Answers

Answers 1

I think this function may help:

  # The graph G is represened by a dictionnary following this pattern:   # G = { vertex: [ (successor1: weight1), (successor2: weight2),...   ]  }   def progress ( G, start ):      Q = [ start ] # contain tasks to execute      done = [ ]    # contain executed tasks      while len (Q) > 0: # still there tasks to execute ?         task = Q.pop(0) # pick up the oldest one          ready = True         for T in G:     # make sure all predecessors are executed            for S, w in G[T]:               if S == task and and S not in done:# found not executed predecessor                   ready = False                  break            if not ready : break         if not ready:            Q.appen(task) # the task is not ready for execution         else:            done.appen(task) # execute the task            for S, w in G[task]:# and explore all its successors               Q.append(S) 

Read More

How to handle file paths in distributed environment

I'm working on setting up a distributed celery environment to do OCR on PDF files. I have about 3M PDFs and OCR is CPU-bound so the idea is to create a cluster of servers to process the OCR.

As I'm writing my task, I've got something like this:

@app.task def do_ocr(pk, file_path):     content = run_tesseract_command(file_path)     item = Document.objects.get(pk=pk)     item.content = ocr_content     item.save() 

The question I have what the best way is to make the file_path work in a distributed environment. How do people usually handle this? Right now all my files simply live in a simple directory on one of our servers.

3 Answers

Answers 1

Well, there are multiple ways to handle it, but let's stick to one of the simpliest one:

• since you'd like to process big amount of files using multiple servers, my first suggestion would be to use the same OS in each server, so you won't have to worry about cross-platform compatibility
• using the word 'cluster' indicates that all of those servers should know their mutual state - it adds complexity, try to switch to the farm of stateless workers (by 'stateless' I mean "not knowing about other's" as they should be aware of at least their own state, e.g.: IDLE, IN_PROGRESS, QUEUE_FULL or more if needed)
• for the file list processing part you could use pull or push model:
  • push model could be easily implemented by a simple app that crawls the files and dispatches them (e.g.: over SCP, FTP, whatever) to a set of available servers; servers can monitor their local directories for changes and pick up new files to process; it's also very easy to scale - just spin up more servers and update the push client (even in runtime); the only limit is your push client's performance
  • pull model is a little bit more tricky, cause you have to handle more complexity; having a set of servers implicates having a proper starting index per node and offset - it will make error handling more difficult, plus, it doesn't scale easily (imagine adding twice as more servers to speedup the processing and updating indices and offsets properly on each node.. seems like an error-prone solution)
• I assume that the network traffic isn't a big concern - having 3M files to process will generate it somewhere, one way or the other..
• collecting/storing the results is a different ballpark, but here the list of possible solutions is limitless

Answers 2

If your are in linux environment the easiest way is mount a remote filesystem, using sshfs, in the /mnt folder foreach node in cluster. Then you can pass the node name to do_ocr function and work as all data is local to current node

For example, your cluster has N nodes named: node1, ... ,nodeN
Let's configure node1, foreach node mount remote filesystem. Here's a sample node1's /etc/fstab file

sshfs#user@node2:/var/your/app/pdfs    /mnt/node2 fuse    port=<port>,defaults,user,noauto,uid=1000,gid=1000        0       0 .... sshfs#user@nodeN:/var/your/app/pdfs    /mnt/nodeN fuse    port=<port>,defaults,user,noauto,uid=1000,gid=1000        0       0 

In current node (node1) create a symlink named as current server pointing to pdf's path

ln -s /var/your/app/pdfs node1 

Your mnt folder should contain remote's filesystem and a symlink

user@node1:/mnt$ ls -lsa 0 lrwxrwxrwx  1 user user      16 apr 12  2016 node1 -> /var/your/app/pdfs 0 lrwxrwxrwx  1 user user      16 apr 12  2016 node2 ... 0 lrwxrwxrwx  1 user user      16 apr 12  2016 nodeN 

Then your function should look like this:

import os MOUNT_POINT = '/mtn' @app.task def do_ocr(pk, node_name, file_path):     content = run_tesseract_command(os.path.join(MOUNT_POINT,node_name,file_path))     item = Document.objects.get(pk=pk)     item.content = ocr_content     item.save() 

It works like all files are in the current machine but there's remote-logic working for you transparently

Answers 3

Since I miss a lot of your architecture details and your application specifics, you can take this answer as a guiding answer rather than a strict one. You can take this approach, in the following order:

1- deploy an internal file server that stores all the files in one place and serve them

Example:

http://interanal-ip-address/storage/filenameA.pdf

http://interanal-ip-address/storage/filenameB.pdf

http://interanal-ip-address/storage/filenameC.pdf

and so on ...

2- Install/Deploy Redis

3- Create an upload client/service/process that takes the files you want to upload and pass them to the above storage location (/storage/), so your files will be available once they are uploaded, at the same time push the full file path URL to a predefined Redis List/Queue (build on linked lists data structure), like this: http://internal-ip-address/storage/filenameA.pdf

You can get more details here about LPUSH and RPOP under Redis Lists here: http://redis.io/topics/data-types-intro

Examples:

1. A file upload form, that stores the files directly to storage area
2. A file upload utility/command-line/background-process, that you can create it yourself or use some existing tool to upload files to the storage location, that gets the files from specific location, be it a web address or some other server that has your files

4- Now we come to your celery workers, each one of your workers should pull (RPOP) one of the files URLs from Redis queue, download the file from your internal file server (we built in first step), and do the required processing on the way you wanted it to be.

An important thing to note from Redis documentation:

Lists have a special feature that make them suitable to implement queues, and in general as a building block for inter process communication systems: blocking operations.

However it is possible that sometimes the list is empty and there is nothing to process, so RPOP just returns NULL. In this case a consumer is forced to wait some time and retry again with RPOP. This is called polling, and is not a good idea in this context because it has several drawbacks

So Redis implements commands called BRPOP and BLPOP which are versions of RPOP and LPOP able to block if the list is empty: they'll return to the caller only when a new element is added to the list, or when a user-specified timeout is reached.

Let me know if that answers your question.

Things to keep in mind

• You can add as many workers as you want since this solution is very scalable, and your only bottleneck is Redis server, which you can make cluster of and persist your queue in case of power outage or server crash

• You can replace redis with RabbitMQ, Beanstalk, Kafka, or any other queuing/messaging system, but Redis has ben nominated in this race due to simplicity and meriad of features introduced out of the box.

Read More