python - uWSGI 和 joblib 信号量 : Joblib will operate in serial mode

我在 Docker 容器内的 Flask 应用程序中运行 joblib，以及由 supervisord 启动的 uWSGI(从启用线程开始)。

网络服务器的启动显示以下错误:

unable to load configuration from from multiprocessing.semaphore_tracker import main;main(15)
/usr/local/lib/python3.5/dist-packages/sklearn/externals/joblib/_multiprocessing_helpers.py:38: UserWarning:

[Errno 32] Broken pipe.  joblib will operate in serial mode

知道如何解决这个问题并使 joblib 并行运行吗？谢谢!

docker 容器中安装了以下包:

pytest==4.0.1
pytest-cov==2.6.0
flake8==3.6.0
Cython==0.29.3
numpy==1.16.1
pandas==0.24.0
scikit-learn==0.20.2
fancyimpute==0.4.2
scikit-garden==0.1.3
category_encoders==1.3.0
boto3==1.9.86
joblib==0.13.1
dash==0.37.0
dash-renderer==0.18.0
dash-core-components==0.43.1
dash-table==3.4.0
dash-html-components==0.13.5
dash-auth==1.3.2
Flask-Caching==1.4.0
plotly==3.6.1
APScheduler==3.5.3

编辑

问题要么是由于 uWSGI、nginx 或 supervisord。 dev/shm 缺少权限不是问题，因为如果我直接运行 Flask 服务器，可以创建信号量。在下面找到三个服务的配置文件。免责声明，我是网络服务器新手，配置是通过从不同的博客复制和粘贴而产生的，只是为了使其工作:-D

所以这是我的 uwsgi 配置:

[uwsgi]
module = prism_dash_frontend.__main__
callable = server

uid = nginx
gid = nginx

plugins = python3

socket = /tmp/uwsgi.sock
chown-socket = nginx:nginx
chmod-socket = 664

# set cheaper algorithm to use, if not set default will be used
cheaper-algo = spare

# minimum number of workers to keep at all times
cheaper = 3

# number of workers to spawn at startup
cheaper-initial = 5

# maximum number of workers that can be spawned
workers = 5

# how many workers should be spawned at a time
cheaper-step = 1
processes = 5

die-on-term = true
enable-threads = true

nginx 配置:

# based on default config of nginx 1.12.1
# Define the user that will own and run the Nginx server
user nginx;
# Define the number of worker processes; recommended value is the number of
# cores that are being used by your server
# auto will default to number of vcpus/cores
worker_processes auto;

# altering default pid file location
pid /tmp/nginx.pid;

# turn off daemon mode to be watched by supervisord
daemon off;

# Enables the use of JIT for regular expressions to speed-up their processing.
pcre_jit on;

# Define the location on the file system of the error log, plus the minimum
# severity to log messages for
error_log /var/log/nginx/error.log warn;

# events block defines the parameters that affect connection processing.
events {
    # Define the maximum number of simultaneous connections that can be opened by a worker process
    worker_connections  1024;
}


# http block defines the parameters for how NGINX should handle HTTP web traffic
http {
    # Include the file defining the list of file types that are supported by NGINX
    include /etc/nginx/mime.types;
    # Define the default file type that is returned to the user
    default_type text/html;

    # Don't tell nginx version to clients.
    server_tokens off;

    # Specifies the maximum accepted body size of a client request, as
    # indicated by the request header Content-Length. If the stated content
    # length is greater than this size, then the client receives the HTTP
    # error code 413. Set to 0 to disable.
    client_max_body_size 0;

    # Define the format of log messages.
    log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
                        '$status $body_bytes_sent "$http_referer" '
                        '"$http_user_agent" "$http_x_forwarded_for"';

    # Define the location of the log of access attempts to NGINX
    access_log /var/log/nginx/access.log  main;

    # Define the parameters to optimize the delivery of static content
    sendfile       on;
    tcp_nopush     on;
    tcp_nodelay    on;

    # Define the timeout value for keep-alive connections with the client
    keepalive_timeout  65;

    # Define the usage of the gzip compression algorithm to reduce the amount of data to transmit
    #gzip  on;

    # Include additional parameters for virtual host(s)/server(s)
    include /etc/nginx/conf.d/*.conf;
}

监督配置:

[supervisord]
nodaemon=true

[program:uwsgi]
command=/usr/bin/uwsgi --ini /etc/uwsgi/uwsgi.ini
stdout_logfile=/dev/stdout
stdout_logfile_maxbytes=0
stderr_logfile=/dev/stderr
stderr_logfile_maxbytes=0

[program:nginx]
command=/usr/sbin/nginx
stdout_logfile=/dev/stdout
stdout_logfile_maxbytes=0
stderr_logfile=/dev/stderr
stderr_logfile_maxbytes=0

第二次编辑

从 Python 3.5 迁移到 3.7.2 后，错误的性质略有变化:

unable to load configuration from from multiprocessing.semaphore_tracker import main;main(15)
/usr/local/lib/python3.7/multiprocessing/semaphore_tracker.py:55: UserWarning:

semaphore_tracker: process died unexpectedly, relaunching.  Some semaphores might leak.

unable to load configuration from from multiprocessing.semaphore_tracker import main;main(15)

非常感谢帮助，这对我来说目前是一个很大的障碍:-/

第三次编辑:

HERE on my github account 是最小、完整且可验证的示例。

您可以通过以下方式轻松运行它make build 后跟 make run 。

它将显示以下日志消息:

unable to load configuration from from multiprocessing.semaphore_tracker import main;main(14)

并在您访问 http://127.0.0.1:8080/ 并出现以下错误时崩溃:

exception calling callback for <Future at 0x7fbc520c7eb8 state=finished raised TerminatedWorkerError>
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/joblib/externals/loky/_base.py", line 625, in _invoke_callbacks
    callback(self)
  File "/usr/local/lib/python3.7/site-packages/joblib/parallel.py", line 309, in __call__
    self.parallel.dispatch_next()
  File "/usr/local/lib/python3.7/site-packages/joblib/parallel.py", line 731, in dispatch_next
    if not self.dispatch_one_batch(self._original_iterator):
  File "/usr/local/lib/python3.7/site-packages/joblib/parallel.py", line 759, in dispatch_one_batch
    self._dispatch(tasks)
  File "/usr/local/lib/python3.7/site-packages/joblib/parallel.py", line 716, in _dispatch
    job = self._backend.apply_async(batch, callback=cb)
  File "/usr/local/lib/python3.7/site-packages/joblib/_parallel_backends.py", line 510, in apply_async
    future = self._workers.submit(SafeFunction(func))
  File "/usr/local/lib/python3.7/site-packages/joblib/externals/loky/reusable_executor.py", line 151, in submit
    fn, *args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/joblib/externals/loky/process_executor.py", line 1022, in submit
    raise self._flags.broken
joblib.externals.loky.process_executor.TerminatedWorkerError: A worker process managed by the executor was unexpectedly terminated. This could be caused by a segmentation fault while calling the function or by an excessive memory usage causing the Operating System to kill the worker. The exit codes of the workers are {EXIT(1), EXIT(1), EXIT(1), EXIT(1)}

最佳答案

这真是一个兔子洞。

Github 上的 joblib 问题页面上有类似的 joblib 与 Uwsgi 失败的帖子。但大多数用于较旧的 multiprocessing 后端。新的 loky 后端应该解决这些问题。
multiprocessing 后端有 PR 为 uwsgi 解决了这个问题:

joblib.Parallel(n_jobs=4,backend="multiprocessing")(joblib.delayed(sqrt)(i ** 2) for i in range(10))

但它有时会随机失败并退回到上述 PR 试图解决的相同问题。

进一步挖掘显示，当前后端 loky 默认情况下在进程上并行化( docs )。但是这些进程没有共享内存访问，因此需要序列化和排队的 channel 。这可能就是 uWSGI 失败而 gunicorn 起作用的原因。

所以我尝试切换到线程而不是进程:

joblib.Parallel(n_jobs=4,prefer="threads")(joblib.delayed(sqrt)(i ** 2) for i in range(10))

它有效:)

关于python - uWSGI 和 joblib 信号量 : Joblib will operate in serial mode，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/54751297/