1. A Small Requirement

We run into small requirements all the time, but they are not always simple to implement. Here is a simple file upload requirement, broken into the following steps:

1. The user uploads a large file on the page
2. The large file is temporarily stored in the internal Ceph
3. A background task downloads the large file from Ceph into Docker
4. A background task uploads the large file from Docker to the external COS

The backend uses Django and is deployed as multiple Docker instances. Multiple instances make scaling out easier and improve the service’s concurrency, but they require the instances to be stateless — the stateful parts must be stored in third-party services, and Ceph is one of them.

Uploading the file directly from the local machine to COS would cause a file that is being uploaded to be lost when a new release is deployed.

2. A Bug Triggered by a Large File

During testing, we found that small files uploaded fine, but uploading files larger than 300MB always failed. The log looked like this:

1
2
3
4
5

[2018-10-23 10:11:18] celery: [2018-10-23 10:11:18,114: ERROR/MainProcess] Process 'Worker-20' pid:45 exited with 'signal 9 (SIGKILL)' [time_stamp=2018-10-23 10:11:18,114, worker=MainProcess, levelname=ERROR]
[2018-10-23 10:11:18] celery: [2018-10-23 10:11:18,206: ERROR/MainProcess] Pool callback raised exception: OperationalError(2006, "MySQL server has gone away (error(32, 'Broken pipe'))") [time_stamp=2018-10-23 10:11:18,206, worker=MainProcess, levelname=ERROR]
[2018-10-23 10:11:18] celery: self._execute_command(COMMAND.COM_QUERY, sql)
[2018-10-23 10:11:18] celery: File "/app/.heroku/python/lib/python2.7/site-packages/pymysql/connections.py", line 970, in _execute_command
[2018-10-23 10:11:18] celery: Traceback (most recent call last):

Package versions in use:

1
2
3

Django==1.8.3
celery==3.1.18
django-celery==3.1.16

There are two errors in the log: one is a process being killed, and the other is the database losing its connection. At first we focused on MySQL server has gone away, but we never managed to fix the problem.

In the end, while looking at the memory usage of the Celery Worker in Grafana, we noticed that every time a large file was uploaded, memory usage spiked sharply and then dropped sharply again. It turned out that memory usage had exceeded the limit and the process was force-killed.

Finally, we solved the problem by optimizing memory usage.

3. A Piece of Optimized Code

Before optimization:

1
2
3
4
5

r = requests.get(self.local_file.url,
                 allow_redirects=True,
                 stream=True,
                 timeout=300)
open(self.local_path, 'wb+').write(r.content)

After optimization:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

if not os.path.exists(os.path.dirname(self.local_path)):
    try:
        os.makedirs(os.path.dirname(self.local_path))
    except OSError as exc:
        if exc.errno != errno.EEXIST:
            raise
    r = requests.get(self.local_file.url,
                     allow_redirects=True,
                     stream=True,
                     timeout=300)
with open(self.local_path, 'wb') as f:
        for chunk in r.iter_content(chunk_size=1024 * 512):
            if chunk:
                f.write(chunk)

Before optimization, Celery held the entire file in memory, and memory usage skyrocketed. After optimization, it first requests the response headers and then reads the file content from the response body chunk by chunk through an iterator. This saves a great deal of memory.

4. A Database Connection Anomaly

Finding the problem and fixing it was not the end of it. There was a strange question here: why did MySQL server has gone away appear at all?

A colleague had run into a similar problem before, where Celery multi-process tasks threw all kinds of database exceptions. The analysis was as follows:

When the Celery Worker starts, djcelery performs DB operations and the database connection is initialized.
After the child process is forked, because it fully copies the parent process's memory data, all Workers share the same MySQL connection (the same socket file). Due to the persistent connections feature, the database connection is never closed. This is a pitfall of the djcelery library combined with multi-process deployment.

The solution was:

Do not disable the persistent connections feature; instead, listen for the signals that mark child process initialization completion and task start. On receiving those signals, manually force-close the Django ORM connections in the current process.

The relevant implementation code is as follows:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from django.db import connections

@signals.task_prerun.connect
def task_prerun(**kwargs):
    for conn in connections.all():
        conn.close()

@signals.worker_process_init.connect
def worker_init(**kwargs):
    for conn in connections.all():
        conn.close()

In fact, here the signal 9 (SIGKILL) and MySQL server has gone away were not thrown by the same Celery Worker. Because they inherit from the same parent process and connection pool, when one child process is killed, another process that is handling a task also runs into trouble.

5. A Related Piece of Django Code

Under high concurrency, frequently creating and closing database connections is inefficient. Django’s persistent connections (long-lived connections) exist precisely to solve this problem.

The principle behind Django’s persistent database connections is that after each database connection is created, the connection instance is kept in a Theard.local instance. On every database request, Django looks up an available connection in the local and reuses it if there is one. A connection is only closed when an exception occurs or when it has existed longer than CONN_MAX_AGE.

The CONN_MAX_AGE parameter can be configured in the settings.py file:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.mysql',
        'NAME': 'mydb',  # 数据库名称
        'USER': 'root',  # 数据库用户名
        'PASSWORD': '',  # 数据库密码
        'HOST': 'localhost',  # 数据库主机，默认为 localhost
        'PORT': '3306',  # 数据库端口
        'CONN_MAX_AGE': 60,  # 0 表示使用完马上关闭，None 表示不关闭
    }
}

Let’s look at how Django manages persistent connections:

django/db/__init_.py

1
2
3

def close_old_connections(**kwargs):
    for conn in connections.all():
        conn.close_if_unusable_or_obsolete()

django/db/backends/base/base.py

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

def close_if_unusable_or_obsolete(self):
    if self.connection is not None:
        # If the application didn't restore the original autocommit setting,
        # don't take chances, drop the connection.
        if self.get_autocommit() != self.settings_dict['AUTOCOMMIT']:
            self.close()
            return

        # If an exception other than DataError or IntegrityError occurred
        # since the last commit / rollback, check if the connection works.
        if self.errors_occurred:
            if self.is_usable():
                self.errors_occurred = False
            else:
                self.close()
                return

        if self.close_at is not None and time.time() >= self.close_at:
            self.close()
            return

django/db/backends/mysql/base.py

1
2
3
4
5
6
7

def is_usable(self):
    try:
        self.connection.ping()
    except Database.Error:
        return False
    else:
        return True

6. References

• https://github.com/celery/django-celery/issues/359