start_background_task

[shitone]

Hi Miguel,

really awesome project.

I have a problem when use the Flask-SocketIO. I wanna start a background task automatic when the socketio server begins to run. The background task can continuously broadcast the message to client by using emit(). But the socketio.start_background_task() is not a decorator, it seems cannot start the background task with the socketio server starting. Moreover, I have a requirement which is starting some background tasks in right time.
Are you planning to become the method start_background_task() to a decorator, and integrate a APScheduler compatibility?

Thanks!
Keep up the good work!

[miguelgrinberg]

Can you just start your background task(s) independently of the server? I don't understand why you want to associate those tasks with the starting of the server, just start the server and your task at the same time.

[shitone]

Thank you for answering my question! The task have to continuously broadcast important messages to all Web clients by using Flask-SocketIO, if I start my tasks independently of the flask server , the task thread is independent from flask server thread, it seems that the message could not reach web clients by using the method socketio.emit(). The job of the task is listening a stream port and when it receives the stream from outside, it will send it to all web clients. In your examples, starting a task needs web user to connect firstly. Like this :

@socketio.on('connect', namespace='/test')
def test_connect():
    global thread
    if thread is None:
        thread = socketio.start_background_task(target=background_thread)
    emit('my_response', {'data': 'Connected', 'count': 0})

In my code, even though there is no web user connecting after staring server, the task thread have to run.

[miguelgrinberg]

Can you start the thread in the global scope, after the socketio object is created? That should not cause any problems from the Socket.IO side.

[shitone]

I use the flask-apscheduler to start the task thread add() like this:

----------------------------------------------------------------
webapp\__init__.py

from flask import Flask
from flask_moment import Moment
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager
from config import Config
from flask_socketio import SocketIO
from flask_apscheduler import APScheduler
from flask_socketio import emit

moment = Moment()
db = SQLAlchemy()
login_manager = LoginManager()
login_manager.session_protection = 'strong'
login_manager.login_view = 'main.login'
socketio = SocketIO()
scheduler = APScheduler()

class Config1(object):
    JOBS = [
        {
            'id': 'job1',
            'func': 'webapp:add',
            'trigger': 'date'
        }
    ]
    SCHEDULER_API_ENABLED = True

def add():
    while(True):
        emit('mytask', {'data': 'yourtask'}, namespace='/awspqc')
        socketio.sleep(5)

def create_app():
    app = Flask(__name__)
    app.config.from_object(Config)
    app.config.from_object(Config1())
    Config.init_app(app)
    moment.init_app(app)
    db.init_app(app)
    login_manager.init_app(app)
    from .main import main as main_blueprint
    from .page1 import page1 as page1_blueprint
    from .page2 import page2 as page2_blueprint

    app.register_blueprint(main_blueprint)
    app.register_blueprint(page1_blueprint, url_prefix='/page1')
    app.register_blueprint(page2_blueprint, url_prefix='/page2')
    socketio.init_app(app, async_mode='eventlet')
    scheduler.init_app(app)
    scheduler.start()
    return app
---------------------------------------------------------
manage.py

from webapp import create_app, socketio
app = create_app()
if __name__ == '__main__':
    socketio.run(app, host='0.0.0.0')
----------------------------------------------------------
test.html
<script type="text/javascript">
            var socket = io.connect('http://' + document.domain + ':' + location.port + '/awspqc');
            socket.on('mytask', function(msg){
                alert("22222");
            });
</script>
------------------------------------------------------------

But, the web clients cannot receive the message from add().

[miguelgrinberg]

The problem is that apscheduler is going to start a thread, and you can't use threads when you work with eventlet. I have never used apshceduler with eventlet, but if you want to use both together, your only hope is that apscheduler is compatible with eventlet's monkey patching. See http://eventlet.net/doc/patching.html for info on monkey patching.

[shitone]

I tried to add mokey_patch() ahead of manage.py like this:

import eventlet
eventlet.monkey_patch()
from webapp import create_app, socketio
from flask_socketio import SocketIO

app = create_app()

if __name__ == '__main__':
    socketio.run(app, host='0.0.0.0')

then the server started
(6736) wsgi starting up on http://0.0.0.0:5000
but the web clients cannot visit the website. It seems that the website is blocking.

[jwg4]

Another approach could be to run the background task independently of the flask server, and to have the task access a HTTP endpoint on the flask server in order to emit a message to the clients.

We wanted to do something like this and have other processes inject messages which would go to all the front end socketio clients. We quickly realized that it was much better to have Flask be the only thing which could emit, and have other processes hit an endpoint which calls emit() in the Flask code.

[shitone]

If the Flask-SocketIO project can emit message to all clients by other processes, that's a very useful job. I have reviewed issuses of Flask-SOketIO project, I found many issuses are also talking about that emiting message by other processes or threads. When would the project update about that?

[miguelgrinberg]

@shitone emitting from an external process is covered in the docs: https://flask-socketio.readthedocs.io/en/latest/#emitting-from-an-external-process

[shitone]

@miguelgrinberg As the docs, the socketio server send messages to Redis or RMQ, but I don't understand that web clients how to receive those messages from Redis or RMQ through websockets.

[miguelgrinberg]

@shitone No, that's not how it works. You have the Flask-SocketIO server and you have another process. The Redis or RabiitMQ queue is in between the two. When you call emit() on the external process, a message is posted to the queue. On the other side, Flask-SocketIO is watching the queue and sees the new message. So it goes and grabs it, and runs the emit, which now reaches the client.

[shitone]

@miguelgrinberg Oh, got it, thanks! I have studied about Celery with Flask-SocketIO. I find an example code combining Celery and Flask-SocketIO like this:

----------------------------------------init.py-----------------------------------
from flask import Flask, render_template                                                         
from flask_socketio import SocketIO                                                              
from celery import Celery                                                                        
import eventlet                                                                                  
eventlet.monkey_patch()                                                                          
                                                                                                 
app = Flask(__name__)                                                                            
here = os.path.abspath(os.path.dirname(__file__))                   
app.config.from_pyfile(os.path.join(here, 'proj/celeryconfig.py'))                               
                                                                                                 
SOCKETIO_REDIS_URL = app.config['CELERY_RESULT_BACKEND']                                         
socketio = SocketIO( app,  async_mode='eventlet',  message_queue=SOCKETIO_REDIS_URL) 
                                                                                                 
celery = Celery(app.name)                                                                        
celery.conf.update(app.config)
-------------------------------------task.py-------------------------------------------
@celery.task                                                                                     
def background_task():                                                                           
    socketio.emit(                                              
        'my response', {'data': 'Task starting ...'},                                            
        namespace='/task')                                                                       
    time.sleep(10)                                                                               
    socketio.emit(                                                                               
        'my response', {'data': 'Task complete!'},                                               
        namespace='/task') 
----------------------------------------view.py---------------------------------------
@app.route('/task')                                                                              
def start_background_task():                                                                     
    background_task.delay()                                                                      
    return 'Started' 
----------------------------------------manage.py----------------------------------
if __name__ == '__main__':
    socketio.run(app, host='0.0.0.0', port=9000, debug=True)
------------------------------------------------------------------------------------

In the example, the celery task(like as another process) and the Flask-IO server use the same instance socketio = SocketIO(). It seems that the same socketio not only posts the message to the queue, but also is watching the queue. Do I understand right ? And could the Flask-SocketIO server and another process share one SocketIO()?

[miguelgrinberg]

There is one difference. In the Flask-SocketIO server, you initialize socket.io as follows:

socketio = SocketIO(app, message_queue="redis://')

while on the Celery worker, you have to do this instead:

socketio = SocketIO(message_queue="redis://")

When the SocketIO object is initialize without a Flask app instance, it creates itself as a write-only object. A server will not be created, and anything you emit on that object will go to to the message queue, while will then be picked up by the Flask-SocketIO server and sent to the client.

[shitone]

In other words, if I want to use emit() with Celery, I have to initialize two SocketIO objects in my app. One object which used to watch the queue is initialized with app, another object which used to emit messages by Celery is initialized without app.

[miguelgrinberg]

@shitone I think the way you are putting it is a bit confusing, at least to me. You have to have only one SocketIO object per process. The processes that are servers (which can be multiple ones) will need the object initialized with the Flask app instance. The processes that are Celery workers (which can also be several) will have the object initialized without a Flask app instance.

[shitone]

@miguelgrinberg Today I setup RabbitMQ add message_queue to SocketIO() like this:

----------------------------------------------------------------
webapp\__init__.py

from flask import Flask
from flask_moment import Moment
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager
from config import Config
from flask_socketio import SocketIO

moment = Moment()
db = SQLAlchemy()
login_manager = LoginManager()
login_manager.session_protection = 'strong'
login_manager.login_view = 'main.login'
socketio = SocketIO()

def create_app():
    app = Flask(__name__)
    app.config.from_object(Config)
    Config.init_app(app)
    moment.init_app(app)
    db.init_app(app)
    login_manager.init_app(app)
    from .main import main as main_blueprint
    from .page1 import page1 as page1_blueprint
    from .page2 import page2 as page2_blueprint

    app.register_blueprint(main_blueprint)
    app.register_blueprint(page1_blueprint, url_prefix='/page1')
    app.register_blueprint(page2_blueprint, url_prefix='/page2')
    socketio.init_app(app , async_mode='eventlet', message_queue='amqp://xxzx:123@10.116.32.197:5672/socketio')
    return app
--------------------------------------------------------

But when I visit the website, the webpage is blocking. If I delete the parameter message_queue, the website recovery。Do you known what is the reason? The blocking webpages all contain websockets JS codes like var socket = io.connect(); socket.on();. It seems that the websocket connection make webpages or flask-server blocking. How can I solve the problem?

[miguelgrinberg]

Are you using eventlet or gevent? If you are, then you need to monkey patch the standard library for rabbitmq to work.

[shitone]

@miguelgrinberg I have used eventlet like this:
socketio.init_app(app , async_mode='eventlet', message_queue='amqp://xxzx:123@10.116.32.197:5672/socketio')
Now I add monkey path to the manage.py like this:

import eventlet
eventlet.monkey_patch()

from webapp import create_app, socketio

app = create_app()

if __name__ == '__main__':
    socketio.run(app, host='0.0.0.0')

But after I add monkey patch, even the login webpage is blocking firstly.

[miguelgrinberg]

Are you sure you have a RabbitMQ service at the address you are using? Do you see the connection from the Rabbit side?

[shitone]

@miguelgrinberg Yes, I have. I cut some pictures of RMQ to you.

[miguelgrinberg]

Okay, so Rabbit is okay. Is the web server running? Does the http://localhost:5000 request reaches the server? You can find out by adding a print statement on the main handler, for example. If it does reach it, it would be useful to find out where it hangs.

[shitone]

@miguelgrinberg Now I add a print statement print 'is reach here?' on the main handler of login like this:

@main.route('/login', methods=['GET', 'POST'])
def login():
    print 'is reach here?'
    if request.method == 'POST':
        username = request.form['username']
        password_md5 = request.form['password']
        remember = request.form['remember']
        user = User.query.filter_by(username=username).first()
        if user is not None and user.verify_password(password_md5):
            login_user(user, remember)
            return json.dumps(dict(succeed = True))
        return json.dumps(dict(succeed = False))
    return render_template('login.html')

But when I visited the login webpage, the console did not print it.

[miguelgrinberg]

I have to think there is something wrong in your setup, but I can't really tell you. If you have your complete application on GitHub or other public repository, I can take a closer look.

[shitone]

@miguelgrinberg I have pushed my project on my GitHub, the project name is mdep.

[miguelgrinberg]

@shitone thanks. Can you give me the output of pip freeze or else add a requirements.txt file to your repository?

[shitone]

@miguelgrinberg Here is the output of pip freeze.

alembic==0.8.6
amqp==1.4.9
anyjson==0.3.3
APScheduler==3.2.0
argh==0.26.1
backports-abc==0.4
backports.ssl-match-hostname==3.5.0.1
billiard==3.3.0.23
bleach==1.4.3
blinker==1.4
blueprint==3.4.2
Blueprints==2.3.0.2
celery==3.1.25
celery-with-redis==3.0
certifi==2016.2.28
cffi==1.9.1
click==6.7
cryptography==1.7.2
cx-Oracle==5.2.1
Django==1.9.6
dominate==2.2.0
enum-compat==0.0.2
enum34==1.1.6
eventlet==0.21.0
Flask==0.12.1
Flask-APScheduler==1.7.0
Flask-Bootstrap==3.3.5.7
Flask-Celery==2.4.3
Flask-Celery-Helper==1.1.0
Flask-HTTPAuth==3.1.2
Flask-Login==0.3.2
Flask-Mail==0.9.1
Flask-Migrate==1.8.0
Flask-Moment==0.5.1
Flask-PageDown==0.2.1
Flask-Script==2.0.5
Flask-SocketIO==2.8.6
Flask-SQLAlchemy==2.1
Flask-WTF==0.12
ForgeryPy==0.1
funcsigs==1.0.2
futures==3.0.5
gevent==1.1.2
gevent-websocket==0.9.5
greenlet==0.4.12
html5lib==0.9999999
idna==2.2
ipaddress==1.0.18
itsdangerous==0.24
Jinja2==2.9.6
kombu==3.0.37
Mako==1.0.4
Markdown==2.6.6
markdown2==2.3.1
MarkupSafe==1.0
mysql-connector-python==2.1.3
mysqlclient==1.3.7
numpy==1.10.4
paramiko==2.1.1
pathtools==0.1.2
pika==0.10.0
pyasn1==0.2.2
pycparser==2.17
python-dateutil==2.6.0
python-editor==1.0
python-engineio==1.5.1
python-socketio==1.7.4
pytz==2016.4
PyYAML==3.11
redis==2.10.5
requests==2.10.0
singledispatch==3.4.0.3
six==1.10.0
SQLAlchemy==1.0.12
tornado==4.3
tzlocal==1.2.2
vine==1.1.3
visitor==0.1.2
watchdog==0.8.3
Werkzeug==0.12.1
WTForms==2.1

[shitone]

@miguelgrinberg Is there something wrong in my setup? I have debugged two days and searched information online, however, I had no progress.

[miguelgrinberg]

@shitone I think the problem is the mysql package that you are using, which is written in C. That is likely incompatible with eventlet. I recommend you switch to a pure python mysql driver, such as pymysql.

[shitone]

@miguelgrinberg Thks! I have pip uninstall mysqlclient==1.3.7. Now I have tried to use two pure python mysql drivers which are mysql-connector-python provided by Oracle and pymysql. But the problem is still existing.

[shitone]

@miguelgrinberg I concluded four appearances about this problem:

1. When the monkey patch added in manage.py and the socketio initialized with RMQ , we can not visit the login webpage. The login webpage request can't reach the flask web routine.

2. If the manage.py doesn't contain monkey patch and the socketio initialized with RMQ, then we can visit the login webpage and login in. After login in, when we visit webpages contain websocket JavaScript codes, the server will be blocking.

3. If the manage.py doesn't contain monkey patch and the socketio initialized without RMQ, all the webpage are normal.

4. If the manage.py contains monkey patch and the socketio initialized without RMQ, we can not visit the login webpage. The login webpage request can't reach the flask web routine.

[miguelgrinberg]

@shitone the mysql-connector-python package is written in C. That's what hanged for me when I tried your project here. I changed the URL for the database to sqlite:// and that made your login page work (without using RabbitMQ).

The RabbitMQ is another possible blocking problem, if you are using librabbitmq instead of amqp, though your virtualenv contents suggest you are actually using amqp.

[shitone]

@miguelgrinberg I have tried pymysql with monkey patch in manage.py (without RMQ), but I still can't access login page. This problem makes me depressing. I don't use virtualenv now, and my python version is 2.7, and I use PyCharm to develop. I doesn't install librabbitmq. When I tried pymysql, I removed message_queue and the key add is monkey patch.

[miguelgrinberg]

@shitone did you try switching to sqlite as I indicated above? That will tell us if the problem is still MySQL related or not. I was able to get to the login prompt after removing message_queue from the SocketIO constructor and changing the database url to sqlite://.

[shitone]

@miguelgrinberg I try switching to sqlite, but I still could not access login page and the requests cannot reach flask main routine. I have pushed new code to github, including removing message_queue in init.py, adding sqlite:/// in Config.py, monkey patch in manage.py, new requirements.txt and a sqlite database called readline.db.
This is the development console, when I visit login page, it's blocking and no information is printed.

If removing monkey patch, after accessing login page, console will like this:

[miguelgrinberg]

Your new version works fine for me. I only changed the database name, then I did:

$ FLASK_APP=manage.py flask run

And I can then connect and see the login prompt.

[shitone]

@miguelgrinberg Thks！I find the cause of this problem just now. The debug model makes app block. After I remove debug=True from Config.py and manage.py, then I can access to login page. Now my app could work with message_queue and pymysql. Do you know the reason why cannot open debug model?

[miguelgrinberg]

Oh, interesting, I did not try to run the application with debug mode, so I never saw this. But now I do see the problem.

It appears there is some sort of incompatibility between eventlet and watchdog, the package that the reloader uses to detect changes in files. Try removing watchdog, that forces Flask to use another method to detect changed files. That fixed the problem for me.

[jpoutrin]

@miguelgrinberg I have a similar issue.
it seems socketio.server.start_background_task method does not spawn a new thread properly.
it seems to use the basic threading module (I mean python Threading)
It seems to start the listener but also kills the original thread... (does not show in the thread list anymore but when I kill the process, it seems the mail thread start running again - may be a my debugger not retrieving all the information)
my debugger shows that in the initialize method of pubsub_manager [33].

     def initialize(self):
        super(PubSubManager, self).initialize()
        if not self.write_only:
            self.thread = self.server.start_background_task(self._thread)
        self.server.logger.info(self.name + ' backend initialized.')

The issue only shows when using the queues obviously.

I am using Python 3.5 not sure if that's the issue.
socket related versions:
Flask-SocketIO==2.8.6
python-socketio==1.7.4
python-engineio==1.5.2
eventlet==0.19.0 and tried 0.21.0

[miguelgrinberg]

@jpoutrin are you basing your assessment on what you see on your debugger, or is there something not working correctly? Debuggers tend to not work well when used with async frameworks, they need to have specific support for them, so I would not trust a thread list shown by the debugger, when in reality when using an async framework all the threads are "green" threads, which is not the same thing.

[jpoutrin]

@miguelgrinberg well, I cannot see the next logging statement:
self.server.logger.info(self.name + ' backend initialized.')
so I assume something is blocking the execution.
You think I am not enabling the logger correctly?

Not sure about green thread (my python background is not up to that level yet ;) ) - what does that mean?
can you help me in debugging that better than prints ? ;)

[miguelgrinberg]

You are using a message queue, correct? The python package that talks to that queue could be incompatible with eventlet. To make it compatible, you have to "monkey-patch" the python standard library. My guess is that this is what's missing in your case. See http://eventlet.net/doc/patching.html for info on monkey patching.

[jpoutrin]

@miguelgrinberg I had the impression that your code was doing that under the hood.
my example works now.

I limited to:
import eventlet eventlet.monkey_patch(socket=True, select=True)
as the full patching generates errors.

as a side note, I only patched the web server.
my worker remains non patched as using async_mode=threading.

thanks for pin pointing me to the right direction.
If I can suggest the patching should be in the main documentation as probably any solid implementation will need something like that.

other question, which backends is recommended for performance vs scale?
or how is redis compared to amqp?

thanks again for the quick support!

[shitone]

@miguelgrinberg I have a problem about Flask-SQLAlchemy and Kombu. I use Flask-APScheduler to start a background thread in order to receive records from RabbitMQ and insert those records to MySQL. The code likes this:

def listen_awspqc_record():
    record_queue = Queue('awspqc_record')

    def _record_aws(body, message):
        awsarrival = AwsArrival(******)
        db.session.add(awsarrival)
        db.session.commit()
        message.ack()

    with Connection('amqp://xxzx:123456@10.116.32.197:5672/cimiss') as conn:
        with conn.Consumer(record_queue, accept=['json'], callbacks=[_record_aws]) as consumer:
            while True:
                conn.drain_events()

However, if it runs at db.session.add(awsarrival), it won't work! I don't know the reason. I guess the key is app_context(), but I don't know how to use app.app_context() with SQLAlchemy and Kombu callbacks.

[miguelgrinberg]

@shitone do this:

with app.app_context():
    db.session.add(...)
    db.session.commit()

[shitone]

@miguelgrinberg How could I add app to the task python file ( rabbit.py )? When I use from manage import app in rabbit.py, the console will print cannot import createapp as soon as the flask server starting. If I import current_app in rabbit.py and use 'current_app._get_current_object().app_context()' in the func def _record_aws(body, message):, the task will not run.

[miguelgrinberg]

@shitone I'm not familiar with how you start a thread with APScheduler, but when you use plain threads, you can pass app as an argument into the thread. There must be a way to do this with APScheduler, I suppose.

def listen_awspqc_record(app):
    with app.app_context():
        # ...

th = Thread(target=listen_awspqc_record, args=(app._get_current_object(),))
th.start()

[shitone]

Hi，@miguelgrinberg How to emit a message to a specify user by user id or user name. For example, there is an alert about User A, I just only want to send the alert to User A excepts other users.

[miguelgrinberg]

You have to use emit(data, room=sid), where sid is the session id assigned to the user you want to address the message. You can access the sid for each user as request.sid in the connect event for that user, as well as in any events that this emits to the server.

[shitone]

@miguelgrinberg Now I have a background thread to emit alerts to the specify User. The requirement mode of emit likes broadcast mode, but the alert is only broadcast to one specify User. I don't have request to get session id.
I use Flask-Login to mange user login and logout. Do Flask-Login store all sessions? If it stores, how can I get the session id by User ID from sessions storage?

[miguelgrinberg]

This is something that your application needs to manage. Flask-Login does not really participate in this.

For example, on the connect event you can write the sid assigned to your user to your database, so that the background thread can obtain it when it needs it.

[shitone]

@miguelgrinberg It seems that request.sid can only be used in the connect event or other socketio events. If I want to get sid by request.sid when users login, the console will print error. Do you know how to get sid when users login?

[miguelgrinberg]

The sid value is an attribute of the Socket.IO connection. It does not exist outside of that. If you need to use it elsewhere, you have to save it when you get it in the connect event.

[simanacci]

@shitone did you find a solution?