DIY web service with asynchronous queues and parallel execution

Everyone must do their job efficiently and on time. Suppose you need to make a web-based image classification service based on a trained neural network using the caffe library . Nowadays, quality means asynchronous non-blocking calls, the possibility of parallel execution of several tasks with free processor cores, monitoring of job queues ... The RQ library allows you to implement all this in a short time without studying tons of documentation.

We will create a web service on one server, focused on lightly loaded projects and relatively long tasks. Naturally, its use is not limited to these of your neural networks.

Formulation of the problem

The input is a file (for example, a JPEG image). For simplicity, we believe that it has already been placed in a dedicated directory. The output is a string in JSON format. For solidity, we will use standard HTTP result codes.

The web service will implement two HTTP calls (let's call this API):

• / process / [file name] - process the file (pre-load the input file in the selected directory, return the job identifier)
• / result / [task identifier] - get the result (if the result is not ready, return the code 202 "Not ready", if the result is ready - return json, if the task identifier does not exist, return the code 404 "Not found")

Install components in Ubuntu

We will use Flask as the HTTP server . Installation:

If pip is not installed:

sudo apt-get install python-pip
sudo apt-get install --upgrade pip

Actually, installing Flask:

pip install flask

Now you need to install Redis - data warehouse and message broker:

wget http://download.redis.io/redis-stable.tar.gz
tar xvzf redis-stable.tar.gz
cd redis-stable
make
sudo make install

Installing the RQ library (Redis Queue):

pip install rq

To automatically start and configure all components, we will use Supervisor :

sudo apt-get install supervisor

We write our service

It is easy in Flask. Create the deep_service.py file :

#Путь к директории, выделенной под хранение входных файлов
BASEDIR = '/home/sergey/verysecure'
#Импортируем служебные библиотеки
import argparse
import os
import json
#Импортируем только что установленные компоненты нашего сервиса
from flask import Flask
app = Flask(__name__)
from redis import Redis
from rq import Queue
#Импортируем функцию, реализующую наш вычислительный процесс, который будет выполняться асинхронно
#classify.py - модуль, поставляемый с библиотекой машинного обучения caffe
#Естественно, вместо него можно импортировать собственные разработки
from classify import main
#Подключаемся к базе данных Redis
q = Queue(connection=Redis(), default_timeout=3600)
#Реализуем первый вызов нашего API
@app.route('/process/')
def process(file_path):
    full_path = os.path.join(BASEDIR, file_path)    #входной файл лежит в директории BASEDIR
    argv = {'input_file': full_path,
                'gpu': True}
    args = argparse.Namespace(**argv)
    r = q.enqueue_call(main, args=(args,), result_ttl=86400)
    return r.id
#В порядке обмена опытом: ограничим 4-мя цифрами после запятой вещественные числа,
#при сериализации в JSON из массива numpy
def decimal_default(obj):
    if isinstance(obj, float32):
        return round(float(obj), 4)
    else:
        raise TypeError()
#Реализуем второй вызов нашего API
@app.route('/result/')
def result(id):
    try:    
        job = q.fetch_job(id)
        if job.is_finished:
            return json.dumps(job.result, ensure_ascii=False, default=decimal_default)
        else:
            return 'Not ready', 202
    except:
        return "Not found", 404
if __name__ == '__main__':
    app.run()
    #app.run(debug=False, host='0.0.0.0')

Manual start - check how it works

At this point, you can check if our web service is working. Launch Redis:

redis-server

We start one workflow (you can run several of them, for example, by the number of processor cores or by the presence of several video cards, if they are required for data processing). It is better to start the process from the directory in which the computational functions will be launched, in our case this is where classif.py lies :

rq worker

We start the http server:

python deep_service.py

We write the cat.jpg picture into the directory for the input data and execute the service request:

wget 127.0.0.1/process/cat.jpg

In response, we get the job ID. Copy the identifier and execute the second request to the service:

wget 127.0.0.1/result/[идентификатор]

In response, we get a JSON string with weights of the image belonging to IMAGENET categories.
Now it remains to configure the automatic launch of our server components.

Autostart

Setting up a supervisor is probably the hardest part of this journey. Good tutorial on setting up supervisor here .

First of all, you need to understand that supervisor runs each process in its own environment. In most cases of complex calculations, the program that implements them depends on a number of settings, such as paths. These settings are usually stored in /home/usersname/.bashrc

For example, the caffe library of neural network computing and the Python modules to it required to add the following lines to this file:

export PATH=/usr/local/cuda-7.5/bin:$PATH
export LD_LIBRARY_PATH=/usr/local/cuda-7.5/lib64:$LD_LIBRARY_PATH
PKG_CONFIG_PATH=$PKG_CONFIG_PATH:/usr/local/lib/pkgconfig
export PKG_CONFIG_PATH
LD_LIBRARY_PATH=/home/username/caffe/build/lib:$LD_LIBRARY_PATH
export PYTHONPATH="${PYTHONPATH}:/home/username/caffe/python"

Copy these lines to the clipboard!

In the directory / usr / local / bin, create the file deep_worker.sh

#!/bin/bash
cd /home/username/caffe/python
export PATH=/usr/local/cuda-7.5/bin:$PATH
export LD_LIBRARY_PATH=/usr/local/cuda-7.5/lib64:$LD_LIBRARY_PATH
PKG_CONFIG_PATH=$PKG_CONFIG_PATH:/usr/local/lib/pkgconfig
export PKG_CONFIG_PATH
LD_LIBRARY_PATH=/home/username/caffe/build/lib:$LD_LIBRARY_PATH
export PYTHONPATH="${PYTHONPATH}:/home/username/caffe/python"
rq worker

Well, you understand - in the first line we go to the working directory, then we paste the environment variables copied from .bashrc, then we start the process.

In the / usr / local / bin directory, create the deep_flask.sh file

#!/bin/bash
cd /home/username/caffe/python
export PATH=/usr/local/cuda-7.5/bin:$PATH
export LD_LIBRARY_PATH=/usr/local/cuda-7.5/lib64:$LD_LIBRARY_PATH
PKG_CONFIG_PATH=$PKG_CONFIG_PATH:/usr/local/lib/pkgconfig
export PKG_CONFIG_PATH
LD_LIBRARY_PATH=/home/username/caffe/build/lib:$LD_LIBRARY_PATH
export PYTHONPATH="${PYTHONPATH}:/home/username/caffe/python"
python deep_service.py

Again - in the first line we go to the working directory, then paste the environment variables copied from .bashrc, then start our Flask server.

A bit of system administration:

sudo chmod +x /usr/local/bin/deep_flask.sh
sudo chmod +x /usr/local/bin/deep_worker.sh
mkdir /var/log/deepservice

In the /etc/supervisor/conf.d directory, create the deepservice.conf file :

[program:redis]
command=/usr/local/bin/redis-server
autostart=true
autorestart=true
stderr_logfile=/var/log/deepservice/redis.err.log
stdout_logfile=/var/log/deepservice/redis.out.log
[program:worker1]
command=/usr/local/bin/deep_worker.sh
autostart=true
autorestart=true
stderr_logfile=/var/log/deepservice/worker1.err.log
stdout_logfile=/var/log/deepservice/worker1.out.log
user=username
directory=/home/username/caffe/python
[program:flask]
command=/usr/local/bin/deep_flask.sh
autostart=true
autorestart=true
stderr_logfile=/var/log/deepservice/flask.err.log
stdout_logfile=/var/log/deepservice/flask.out.log
user=username
directory=/home/username/caffe/python

Finally, run this whole construct:

sudo supervisorctl reread
sudo supervisorctl update

All!