Убить отдельные темы, когда тайм-аут в службе

Question

Я создаю службу фляги Python, для которой я пытаюсь установить таймаут для каждого отдельного запроса POST. Как я понимаю, всякий раз, когда кто-то отправляет запрос на публикацию в мой сервис RESTful, начинает выполняться новый поток (виртуальный или реальный).

Теперь, чтобы мой сервер обслуживал множество запросов, я хочу, чтобы он возвратилОтвет TIME-OUT, если процесс выполняется более определенного для него постоянного времени (TIMEOUT_TIME), установленного для каждого метода POST, и останавливает выполнение этого отдельного потока.

Можете ли вы предложить мне абстрактную схему, которую я мог быреализовать, используя колбу-методы?

Answer 1

Один из способов сделать это - запустить обработку запроса в отдельном процессе и прекратить его, если превышено время ожидания:

#!/usr/bin/env python3
import time
from multiprocessing import Process

from flask import Flask, request, jsonify


app = Flask(__name__)


@app.route('/api/sleep', methods=['POST'])
def sleep():
    duration = int(request.args.get('duration', 1))
    timeout = float(request.args.get('timeout', 2))

    proc = Process(target=process_request, args=(duration,))
    proc.start()
    proc.join(timeout)

    if proc.is_alive():
        proc.terminate()
        proc.join()

        return jsonify(success=False, message='timeout exceeded'), 408

    return jsonify(success=True, message='well done')


def process_request(t):
    time.sleep(t)


if __name__ == '__main__':
    app.run(host='localhost', port=8080, debug=True)

В этом примере, когда время ожидания duration меньшепри timeout пользователь получит успешный ответ:

curl -X POST http://localhost:8080/api/sleep?duration=1\&timeout=2
{
  "message": "well done", 
  "success": true
}

В противном случае пользователь получит 408 ошибку:

curl -X POST http://localhost:8080/api/sleep?duration=2\&timeout=1
{
  "message": "timeout exceeded", 
  "success": false
}

Проблема с этим подходом отмечена в документы

Обратите внимание, что обработчики выхода и предложения finally и т. д. не будут выполняться.

Это означает, что запущенные процессы выиграли 'не может быть убран перед выходом, что может вызвать проблемы. Другое решение состоит в том, чтобы использовать специальный поток Joiner, который будет использоваться для последующего присоединения к рабочим процессам или потокам в случае превышения времени ожидания:

#!/usr/bin/env python3
import time
from queue import Queue
from threading import Thread

from flask import Flask, request, jsonify


class Joiner(Thread):

    def __init__(self):
        super().__init__()
        self.workers = Queue()

    def run(self):

        while True:
            worker = self.workers.get()

            if worker is None:
                break

            worker.join()


app = Flask(__name__)


@app.route('/api/sleep', methods=['POST'])
def sleep():
    duration = int(request.args.get('duration', 1))
    timeout = int(request.args.get('timeout', 2))

    worker = Thread(target=process_request, args=(duration,))
    worker.start()
    worker.join(timeout)

    if worker.is_alive():
        joiner.workers.put(worker)

        return jsonify(success=False, message='timeout exceeded'), 408

    return jsonify(success=True, message='well done')


def process_request(t):
    time.sleep(t)


if __name__ == '__main__':
    joiner = Joiner()
    joiner.start()

    app.run(host='localhost', port=8080, debug=True)

    joiner.workers.put(None)
    joiner.join()

Здесь, перед запуском колб-сервера a Joiner экземпляр потока создан и запущен. Как только сервер остановлен, мы помещаем None в очередь joiner.workers, чтобы дать сигнал потоку joiner завершиться.