Блог

на моем HTTP-сервере на базе flask, предназначенном для удаленного управления некоторыми сервисами в RPI, я столкнулся с проблемой, которую не могу решить в одиночку, поэтому любезно прошу вас дать мне подсказку.

Концепция: С помощью flask и gevent я могу остановить и запустить некоторые (две) службы, работающие на RPI. Я использую gevent и событие на стороне сервера в отношении javascript, чтобы прослушивать обновления html.

html-страница показывает состояние (включено / выключено / обработка) сервисов и предоставляет кнопки для их включения / выключения. Дополнительно отобразите некоторые системные параметры (CPU, RAM, HDD, NET).

Пока открыт только один пользователь / страница, все работает так, как требуется. Как только появляется больше пользователей, обращающихся к серверу flask, возникает гонка между гринлетами, обслуживающими каждого пользователя / страницу, и не все страницы перезагружаются.

Проблема: Как я могу отправить сообщение всем запущенным гринлетам sse_worker() и обработать его поверх их обычной работы?

Ниже приведен код высокого уровня. Полный исходный код можно найти здесь:https://github.com/petervflocke/flasksse_rpi проверьте sse.py файл

 def sse_worker(): #neverending task
   while True:
       if there_is_a_change_in_process_status:
          reload_page=True
       else: 
          reload_page=False
       Do some other tasks:
          update some_single_parameters_to_be_passed_to_html_page
       yield 'data: '   json.dumps(all_parameters)
       gevent.sleep(1)

@app.route('/stream/', methods=['GET', 'POST'])
def stream():
    return Response(sse_worker(), mimetype="text/event-stream")

if __name__ == "__main__":
    gevent.signal(signal.SIGTERM, stop)
    http_server = WSGIServer(('', 5000), app)
    http_server.serve_forever()
  

…на html-странице потоковые данные json обрабатываются соответствующим образом. Если статус службы был изменен на основе переменной reload_page, javascript перезагрузит полную страницу — выдержка кода ниже:

 <script>
    function listen() {
        var source = new EventSource("/stream/");
        var target1 = document.getElementById("time");
        ....  
        source.onmessage = function(msg) {
          obj = JSON.parse(msg.data);
          target1.innerHTML = obj.time;
        ....
          if (obj.reload == "1") {
            location.reload();
          }
        }
    }
    listen();
</script>
  

Моим желаемым решением было бы расширить sse_worker() следующим образом:

 def sse_worker():
   while True:
       if there_is_a_change_in_process_status:
          reload_page=True
          # NEW: set up a semaphore/flag that there is a change on the page
          message_set(reload)
       elif message_get(block=false)==reload: # NEW: check the semaphore
          # issue: the message_get must retun "reload" for _all_ active sse_workers, that all of them can push the reload to "their" pages 
          reload_page=True
       else: 
          reload_page=False
       Do some other tasks:
          update some_single_parameters_to_be_passed_to_html_page
       yield 'data: '   json.dumps(all_parameters)
       gevent.sleep(1)
  

Я надеюсь, что смог передать свое сообщение. Есть идеи с вашей стороны, как я могу решить проблему синхронизации? Пожалуйста, обратите внимание, что здесь у нас есть производитель и потребитель в одной и той же функции sse_worker.

Любая идея приветствуется! с наилучшими пожеланиями, Питер