Как при асинхронной обработке с помощью нейросетей не подтягивать каждый раз веса?

Question

[Дмитрий Сергеев]

Доброго времени суток, всем!

Столкнулся с проблемой.
Есть нейросеть на базе darknet. Работает отлично и быстро.
Проблема в том, что мы получаем изображение по http и обрабатываем его с помощью python rq, но на каждую задачу он тащит веса.

Вопрос: Как сделать, чтобы веса подтягивались только 1 раз и использовались в последующих асинхронных задачах?

Flask App

import copy
import os

import cv2
import werkzeug
from flask import Flask, flash, request
from redis import Redis
from rq import Queue

from jobs import detect
...
app = Flask(__name__)
q = Queue(connection=Redis())
...

@app.route('/url/', methods=['GET', 'POST'], strict_slashes=False)
def webhook():
    if request.method == 'POST':
        ...
        detect_job = q.enqueue(detect, args=(cam_id, timestamp, file_path)) # распознаем объекты
        ...
    return "", 200

jobs.py

import copy
import urllib

import cv2
from rq import get_current_job

from pydarknet import Image
from utils import write_to_sheet


def detect(spreadsheet_id, timestamp, file_path):
    from detector import detector

    img = Image(cv2.imread(file_path))

    return {
        ...
    }

detector.py

from pydarknet import Detector

detector = Detector(bytes('cfg/yolov3.cfg', encoding='utf-8'),
                    bytes('weights/yolov3.weights', encoding='utf-8'),
                    0,
                    bytes('cfg/coco.data',encoding='utf-8'))

Сейчас код выглядит так. Я пробовал в разные места ставить, но результат один.

Буду благодарен любой наводке, на решение этой задачи!

Comments

[NaName]

что такое "тащит веса"? при каждом запросе надо тренировать новую модель или использовать тренированную модель (всегда одну) для предсказаний?

[Дмитрий Сергеев]

Вот эту штуку тянет в память weights/yolov3.weights, а она 300 мб вешает. Я недавно в этой теме, не очень понимаю, что вообще под капотом происходит.

[NaName]

def detect(spreadsheet_id, timestamp, file_path):
    from detector import detector

что делает этот импорт? что при импорте происходит с detector.detector?)

[Дмитрий Сергеев]

Скорее все нейронка тренеруется и возвращает detector

Answer 1

[Дмитрий Сергеев]

После более глубокого изучения python rq, сделали следующим образом, вместо запуска rq worker, мы запускаем скрипт python run_rq.py. Почти не измененный пример из документации.

# run_rq.py
import sys
from rq import Connection

from worker import CustomWorker

# Preload libraries
from pydarknet import Detector

detector = Detector(bytes('cfg/yolov3.cfg', encoding='utf-8'),
                    bytes('weights/yolov3.weights', encoding='utf-8'),
                    0,
                    bytes('cfg/coco.data',encoding='utf-8'))

# Provide queue names to listen to as arguments to this script,
# similar to rq worker
with Connection():
    qs = sys.argv[1:] or ['default']

    w = CustomWorker(qs)
    w.detector = detector
    w.work()


# worker.py
from rq.worker import (DEFAULT_JOB_MONITORING_INTERVAL, DEFAULT_RESULT_TTL,
                       DEFAULT_WORKER_TTL, Worker)

import settings


class CustomWorker(Worker):
    def __init__(self, queues, name=None, default_result_ttl=DEFAULT_RESULT_TTL,
                 connection=None, exc_handler=None, exception_handlers=None,
                 default_worker_ttl=DEFAULT_WORKER_TTL, job_class=None,
                 queue_class=None,
                 job_monitoring_interval=DEFAULT_JOB_MONITORING_INTERVAL):

        super(CustomWorker, self).__init__(queues, name, default_result_ttl,
            connection, exc_handler, exception_handlers, default_worker_ttl,
            job_class, queue_class,job_monitoring_interval)

        self.detector = None

    def execute_job(self, job, queue):
        if not self.detector is None:
            job.detector = self.detector

        super(CustomWorker, self).execute_job(job, queue)

# tasks.py
from rq import get_current_job

import settings


def send():
    job = get_current_job()
    if not job.detector is None:
        # do more work

Далее внутри кастомного воркера в методе execute_job мы и передаем инстанс detector в внутрь Job.
Таким образом каждый инстанс Job получает уже посчитанный заранее Detector.

Answer 2

[rPman]

Смотрю примеры на github там все выглядит не так как у вас:
https://github.com/pjreddie/darknet/blob/master/ex...

там же прямо в примере показана однократная загрузка нейросети и конфигов а затем последовательно подставляются картинки

Comments

[Дмитрий Сергеев]

Это понятно. Но htpp не может ждать пока у меня все картинки обработаются. Да в таком случае веса загружаются 1 раз, с этим я согласен. Но хочется асинхронно обрабатывать.

[rPman]

А зачем тогда вы используете python rq? если я верно понимаю это библиотека для создания воркеров, позволяющих создавать задачи и исполнять их по очереди в отдельном процессе?

Если вам нужно на каждый http запрос выдать результат - делайте обработку данных прямо в скрипте веб сервера. Лимит на количество одновременно запущенных процессов будет задаваться вебсервером (количество одновременных подключений, при превышении клиенты будут тупо ждать). Поднимите на питоне простейший веб сервер (там 3-5 строчек), а ваш основной веб сервер будет проксировать на него отдельные запросы (либо простым file_get_contents в скрипте).

Answer 3

[Арсений Кравченко]

Сейчас на каждом вызове `detector` инициалируется заново, нужно сделать это один раз в самом начале и замкнуть/каррировать его в функцию, которая передается в таске.

Comments

[Дмитрий Сергеев]

Так не получилось. Ругается на то, что объект не сериализуемый.