Concurrent Request Processing Issue in Flask with Gunicorn and Gevent-Websocket on AWS

[dev-nasir-iqbal]

Hi everyone,

I hope you're doing well.

I'm currently working on a Flask project where I'm utilizing flask_login, flask_sqlalchemy, and flask_socketio. The application is deployed on AWS Elastic Beanstalk using Gunicorn .

web: gunicorn --worker-class geventwebsocket.gunicorn.workers.GeventWebSocketWorker -w 1 --timeout 340 --graceful-timeout 340 application:application

I'm encountering an issue with request processing: when an I/O(database) operation is performed in one http request, other HTTP requests remain in the queue. According to the Flask-SocketIO documentation, increasing the -w (worker count) parameter beyond 1 is not suggested.

Despite the application being load balanced , requests still get stuck in the queue. I understand that Flask operates as a WSGI server, but I am seeking a solution to enable concurrent request processing.

I would greatly appreciate any suggestions or guidance on how to address this issue.

Thank you!

[miguelgrinberg]

According to the Flask-SocketIO documentation, increasing the -w (worker count) parameter beyond 1 is not suggested.

I don't understand how this is related to your problem. You are using gevent, so concurrency happens through greenlets, not worker processes. One Gunicorn worker can spawn up to 1000 greenlets, each handling one client.

I think you need to look at how you run database queries. Whatever you are doing there must be blocking the gevent loop.

[dev-nasir-iqbal]

Thank you for your response. I understand your point regarding Gunicorn workers and concurrency through greenlets.

The issue seems to arise particularly when fetching data from the database, which causes the application to become unresponsive. For instance, when I need to retrieve over 2000+ rows, other HTTP requests remain blocked. Although I've implemented pagination to fetch smaller sets of data, I still want to understand the underlying issue causing the server to become unresponsive during database queries.

Here is an example of a route I was using before implementing pagination:

admin_api.py

@admin_api.get('/all_user')
@login_required
def all_user():
    if current_user.type != "admin" :
        return jsonify('Unauthorized'), 400

    # Fetch all users in a single query
    users = User.query.filter(User.type != "admin").all()

    # Collect all referral IDs
    all_referral_ids = [referral for user in users for referral in user.referrals]

    # Fetch all referrals in a single query
    referrals = User.query.filter(User.id.in_(all_referral_ids)).all()

    # Create a dictionary for quick lookup of referral data
    referral_dict = {referral.id: referral for referral in referrals}

    all_users = [{
        **user.as_dict(),
        'referrals': [referral_dict[referral_id].as_dict() for referral_id in user.referrals if referral_id in referral_dict]
    } for user in users]

    return jsonify(all_users)

init_.py

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager
from flask_session import Session
from flask_cors import CORS
from flask_wtf.csrf import CSRFProtect
import mysql.connector

db = SQLAlchemy()
settings = {}
run_mode = 'local'
csrf = CSRFProtect()

def create_app():
    app = Flask(__name__, static_url_path='/static', static_folder='static')
    app.config["SECRET_KEY"] = config.get('secret_key')
    app.config["SESSION_TYPE"] = "sqlalchemy"
    app.config['SESSION_SQLALCHEMY'] = db
    app.config['SEND_FILE_MAX_AGE_DEFAULT'] = 43200
    Session(app)
    CORS(app)
    csrf.init_app(app)
    db_params = config.get('db_params')
    if run_mode == 'local' or run_mode == 'aws':
        if run_mode == 'local':
            db_params = {
                'user': 'root',
                'password': 'root',
                'host': 'localhost',
                'port': 3307,
                'database': 'uniex_ng'
            }
        user = db_params.get('user')
        pw = db_params.get('password')
        host = db_params.get('host')
        port = db_params.get('port')
        db_name = db_params.get('database')
        app.config['SQLALCHEMY_DATABASE_URI'] = f'mysql+mysqlconnector://{user}:{pw}@{host}:{port}/{db_name}'

    db.init_app(app)

    login_manager = LoginManager()
    login_manager.login_view = 'auth.login'
    login_manager.init_app(app)

    from .models import User

    @login_manager.user_loader
    def load_user(user_id):
        return User.query.get(int(user_id))

    from .routes.admin import admin_api as admin_api_blueprint
    app.register_blueprint(admin_api_blueprint)

    from .routes.socket_events import socketio
    socketio.init_app(app, cors_allowed_origins="*")
    return app, socketio, db

Models.py

from . import db
from flask_login import UserMixin
from sqlalchemy.orm import class_mapper

class User(UserMixin, db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), nullable=False)
    password = db.Column(db.String(200), nullable=False)
    address = db.Column(db.String(120), nullable=False)
    referrals = db.Column(db.JSON, default=[])
    type = db.Column(db.String(20))
    __mapper_args__ = {
        'polymorphic_on': type,
        'polymorphic_identity': 'user'
    }
    def as_dict(self):
        mapper = class_mapper(type(self))
        obj = {
        c.name: ( getattr(self, c.name)
            for c in mapper.columns
        }
        return obj

class Admin(User):
    id = db.Column(db.Integer, db.ForeignKey('user.id'), primary_key=True)
    level = db.Column(db.String(80), nullable=False)
    __mapper_args__ = {
        'polymorphic_identity': 'admin'
    }

class Student(User):
    id = db.Column(db.Integer, db.ForeignKey('user.id'), primary_key=True)
    school_email = db.Column(db.String(80), nullable=False)
    __mapper_args__ = {
        'polymorphic_identity': 'student'
    }

# and so on...

application.py

from flask_migrate import Migrate
from project import create_app

application, socketio, db = create_app()
migrate = Migrate(application, db)
if __name__ == "__main__":
    socketio.run(application, debug=True, host="0.0.0.0")

Could you please provide further insights or suggestions on how to handle this issue more effectively?

Thank you!

[RonnyPfannschmidt]

At first glance it looks like you're using sync SQL alchemy

That is a massive problem for gevent

A workaround I'm aware of is enabling gevent monkey patching

Personally I'd just migrate to async api based on my negative experiences with gevent in the last few years

[miguelgrinberg]

Gunicorn monkey patches by default when using the gevent worker, so I think SQLAlchemy should work fine.

My bet is that mysqlconnector, which is a C library, is incompatible with gevent. I would suggest you switch to pymysql for your database driver, that is more likely to work well with gevent's monkey patching because it is a pure Python library.

[dev-nasir-iqbal]

Thank you for the suggestions, @RonnyPfannschmidt .

Firstly, I did consider shifting to Quart for async API calls because it seemed straightforward. However, I encountered issues with the lack of documentation and library support. My codebase is quite established, and migrating from sync to async would take a considerable amount of time.

@miguelgrinberg , your point about mysqlconnector being a C library and possibly incompatible with gevent makes sense. I will try switching to pymysql as my database driver and see how that works with gevent's monkey patching.

I appreciate both of your inputs and will update on the progress.

Thank you!

[dev-nasir-iqbal]

@miguelgrinberg Thanks Sir, Pymysql work like a charm