• A programmer from game company in Taiwan
• Backend (and temporary frontend) engineer, focus on common service
• Usually develop something related to my work in Python, Ruby, ECMAScript, Golang, C#
• ECMAScript hater since Netscape is dead

• Built CDN-aware game asset update system
• Built almost entire VOEZ game server by myself only
• Supported Sdorica backend development
• Verify and tweak Soul of Eden backend performance

4. Celery
5. Singleton Task
   • Situations
   • Cases
6. Current Implementation
   • Principle
   • Local Mutex
   • Traditional Database
   • Redis & Google Cloud Memory Store

7. Alternative Implementation
   • Repository
   • Inspect
   • Task Collective Suicide
   • Demo
8. Conclusion
9. Reference
10. Q&A

What is Celery

https://docs.celeryproject.org/en/stable/

Structure

digraph {
    compound=true
    rankdir=LR

    graph [fontname="Source Sans Pro" fontsize=20]
    node [fontname="Source Sans Pro" fontsize=18]
    edge [fontname="Source Sans Pro" fontsize=12]

    subgraph cluster_service {
        label="service"

        task_1 [label="task 1" shape=box]
        task_2 [label="task 2" shape=box]
        peroidic_task [label="periodic task" shape=box]
    }

    subgraph cluster_broker {
        label="Celery broker (call, inspect)"
        concentrate=true

        amqp [label="RabbitMQ AMQP" shape=box]
    }

    subgraph cluster_backend {
        label="Celery backend (result)"
        concentrate=true

        rpc [label="RabbitMQ RPC" shape=box]
    }

    subgraph cluster_worker {
        label="Celery worker"

        worker_1 [label="worker 1" shape=box]
        worker_2 [label="worker 2" shape=box]
        worker_3 [label="worker 3" shape=box]
    }

    task_1 -> amqp [color="blue"]
    task_2 -> amqp [color="blue"]
    peroidic_task -> amqp [color="blue"]

    amqp -> worker_1 [color="blue"]
    amqp -> worker_2 [color="blue"]
    amqp -> worker_3 [color="blue"]

    amqp -> worker_1 [color="pink" style="dashed" dir="both"]
    amqp -> worker_2 [color="pink" style="dashed" dir="both"]
    amqp -> worker_3 [color="pink" style="dashed" dir="both"]

    rpc -> task_1 [color="red"]
    rpc -> task_2 [color="red"]
    rpc -> peroidic_task [color="red"]

    worker_1 -> rpc [color="red"]
    worker_2 -> rpc [color="red"]
    worker_3 -> rpc [color="red"]
}

Usage

@app.task
def my_task(arg1, arg2, kwarg1='a', kwarg2='b'):
    ...
    return result
    
result = my_task.delay(arg1, arg2, kwarg1='x', kwarg2='y')
result = my_task.apply_async(args=[arg1, arg2],
    kwargs={'kwarg1': 'x', 'kwarg2': 'y'},
    countdown=60, expires=120)
 
result.get()

Situations

• Only one task can be running at any given time
  • Launch only one worker
    • Pile tasks up
• Only one task can be queued at any given time
  • Skip running task which is queued later

Cases

• Garbage collect any resource
  • Clear blobs on S3/GCS
• Schedule background updating task
  • Update all user data everyday

Principle #1

• Sender checks existence of same task
• If there is no same task queued or running
  • Send task to queue

Principle #2

• Worker accepts task
• If any same task is already running
  • Skip task and quit execution
    • Need at least two workers to flush out tasks

Local Mutex

• Try to lock mutex
• Use mutex locking to identify if task is running

Traditional Database

• Try to write semaphore into database
• Check database to identify if task is running

Redis & Google Cloud Memory Store

• Try to write a lock into Redis
  • Use SETNX
• Check Redis lock to identify if task is running
  • Identify return value of SETNX

Redis Based Example

Celery-Singleton
https://github.com/steinitzu/celery-singleton

Redis Based Feature

• Efficient
• Responsive
• No one will unlock Redis lock if worker is panic
  • Set lock TTL
    • Cannot estimate reasonable TTL when task duration is very unstable

Repository

CCS

CCS

Card Captor Sakura

CCS

Celery-controlled singleton
https://github.com/fieliapm/celery_controlled_singleton

Inspect

• If any same task is already running (active)
  • Skip task and quit execution

Idea

i = celery_app.control.inspect(timeout=1.0) # collect response after 1.0 second
i.active() # get active tasks
# then find same tasks in active tasks

Detail

• inspect.active(): running tasks
• inspect.reserved(): queued tasks
• inspect.scheduled(): scheduled ETA tasks

You cannot atomically inspect all of them

Feature

• Broker/backend independent
  • Based on any broker/backend such as AMQP/RPC
  • Don't need Redis lock
    • Don't need to argue with SRE
• Inspect is slow
  • Affected by timeout parameter
    • Short timeout causes inspect result missing
  • Are delayed tasks really urgent?

Race Condition

Task Collective Suicide

Task Collective Suicide

Reason

• Task 1 and task 2 have same function & parameter
• Worker 1 accepts task 1
• Worker 2 accepts task 2
  • They started simultaneously
• Worker 1 inspects and found task 2
• Worker 2 inspects and found task 1
• Worker 1 skips task 1 and quit
• Worker 2 skips task 2 and quit
  • No worker finishes the singleton task

Solution

• Absolute time_start
  • Since v4.2.0rc1
  • Find same tasks with the earliest time_start

Trap

• Multiple worker instances
  • No truly absolute time_start
    • Race hazard happens

Reason

• Current standard time is 12:00:01
• Worker 1 accepts task 1 at 12:00:03 (12:00:01)
• Worker 1 inspects task 1 is first task
• Worker 2 accepts task 2 at 12:00:00 (12:00:02)
• Worker 2 inspects task 2 is first task
  • Because task 2 has earlier time_start
• Worker 1 runs task 1
• Worker 2 runs task 2
  • Two same tasks are running

Reason (Alternative)

• Worker 1 accepts task 1
• Worker 2 accepts task 2
• Worker 2 reports task 2 active to broker
• Worker 2 inspects task 2 is first task
• Worker 1 reports task 1 active to broker (lag)
• Worker 1 inspects task 1 is first task
• Worker 1 runs task 1
• Worker 2 runs task 2
  • Two same tasks are running

Mitigate Race Condition

• Delay before inspection
  • To ensure all same tasks listed in active list

Demo

Celery-controlled singleton
https://github.com/fieliapm/celery_controlled_singleton

Contribute!

• fieliapm/celery_controlled_singleton
  • Better implementation
    • No race condition

郭學聰 Hsueh-Tsung Kuo2020_09_06

Link

• celery
  • https://docs.celeryproject.org/en/stable/
• how to implement single task:
  • https://stackoverflow.com/questions/20894771/celery-beat-limit-to-single-task-instance-at-a-time
• another possibility:
  • https://stackoverflow.com/questions/20091505/celery-task-with-a-time-start-attribute-in-1970/20096342
  • https://github.com/celery/celery/pull/3684
    https://github.com/celery/celery/blob/c0e587e94575d7be9bfdabe40c1d5dfda268e68e/celery/worker/request.py#L432