client connected then disconnected automatically and frequently on AWS EC2 server

[kashish63]

Hello There,
I am stuck in a condition where I am figuring out what the problem is. Please help me.
I wrote flask socket io code to handle three events. This code works perfectly on my local computer when I am connecting with my localhost client. But when I push that code on AWS EC2 server, it gives me error.
ERROR - server=chatbot.com//socket.io/ client=10.4.1.4:21688 socket shutdown error: [Errno 9] Bad file descriptorserver=chatbot.com//socket.io/ client=10.4.1.5:32876 socket shutdown error: [Errno 9] Bad file descriptorI

After that error server is not behaving the same. It is making multiple connection and then disconnect every 10 seconds.
Evenlet monkey patch also not working on AWS EC2 server.

Dependencies
python 3.10
Flask-SocketIO 5.3.6
requests==2.31.0
flask-cors==4.0.0
eventlet== 0.36.1
gevent=24.2.1
gevent-websocket 0.10.1

my code is

# import eventlet
# eventlet.monkey_patch()   #added
from gevent import monkey
monkey.patch_all()
import gevent

from flask import Flask, request, render_template
from flask_socketio import SocketIO, emit
from utils import openai_fun, functions, active_thread
from database import mongo_database
from flask import jsonify
from flask_cors import CORS,cross_origin
from datetime import datetime    
import logging 
import traceback 
app = Flask(__name__)
CORS(app,supports_credentials=True, resources={r"/*": {"origins": "*"}})
socketio = SocketIO(app,cors_allowed_origins="*", async_mode='gevent', ping_timeout=80, ping_interval=25)

logging.basicConfig(level=logging.INFO)   #added
# CORS(app, resources={r"/*": {"origins": "*"}})

# logging.getLogger('flask_cors').level = logging.DEBUG
clients = {}
 
@socketio.on('connect')
def handle_connect(): 
    client_sid = request.sid
    logging.info(f'Connected: {request.sid}')
    print("connected hua hy ",client_sid) 
    clients[request.sid] = request.sid
@socketio.on('disconnect')
def handle_disconnect(): 
    logging.info(f'Disconnected: {request.sid}')
    if request.sid in clients:
        del clients[request.sid]


@socketio.on_error()        # Handles the default namespace
def error_handler(e):
    print(e)
    pass

@socketio.on_error_default  # handles all namespaces without an explicit error handler
def default_error_handler(e):
    print(e)
    pass

@app.route("/", methods=['GET']) 
def index(): 
    return render_template('index.html')


@socketio.on('gpt_query')
def handle_gpt_query(data):
    try:
 
        question = data.get('question', '').strip()
        thread_id = data.get('thread_id')
        user_id = data.get('user_id', '').strip()
        customerId = data['customerId']
        customerName = data['customerName']
        customerBranchId = data['customerBranchId']
        customerBranchName = data['customerBranchName']
        customerBranchPersonId = data['customerBranchPersonId']
        customerBranchPersonEmail = data['customerBranchPersonEmail']

        credentials = {
            "customerId":customerId, 
            "customerName":customerName,
            "customerBranchId":customerBranchId,
            "customerBranchName":customerBranchName,
            "customerBranchPersonId":customerBranchPersonId,
            "customerBranchPersonEmail":customerBranchPersonEmail
        }
        print(customerId, customerName, customerBranchId, customerBranchName, customerBranchPersonId, customerBranchPersonEmail  )
        if not user_id:
            emit('error_response', {"status": False, "error": "User ID is missing or empty"}, to=request.sid)
            return 
        if not question:  
            emit('error_response', {"status": False, "error": "Question is missing or empty"}, to=request.sid)
            return
        if not thread_id: 
            query_head = question[:40]
            thread = openai_fun.create_thread()
            thread_id = thread.id
            thread_data = { 
                "user_id": user_id,
                "thread_id": thread_id,
                "query_head": query_head,
                "created_at": datetime.now(),
                "isDeleted": False,
            }   
            mongo_database.insert_data(thread_data)  
            logging.info(f"New thread created with ID: {thread_id} for user: {user_id}")

        # response = openai_fun.GPT_Query1(thread_id, question, request.sid)
        openai_fun.GPT_Query2(thread_id, question, request.sid, socketio, credentials) 
        # emit('query_response', {"status": True, "thread_id": thread_id, "response": response}, to=request.sid)

    except KeyError as e: 
        logging.error(f"Missing key: {e.args[0]}")
        emit('error_response', {"status": False, "error": f"Missing key: {e.args[0]}"}, to=request.sid)

    except Exception as e:
        logging.error(f"Unhandled exception: {str(e)}")
        emit('error_response', {"status": False, "error": str(e)}, to=request.sid)
  

@socketio.on('history_massege')
def history_massege(query):
    try:      
        user_id = query.get('user_id', '').strip()
        if not user_id:
            emit('error_response_history', {"status": False, "error": "User ID is missing or empty"}, to=request.sid)
            return          
        thread_id = query.get('thread_id', '').strip()    
        if not thread_id:
            emit('error_response_history', {"status": False, "error": "Thread Id is missing or empty"}, to=request.sid)
            return 
        flag = mongo_database.checkDB(user_id, thread_id)
        if(flag==True): 
            messageList = openai_fun.previous_message(thread_id)
            messages = []    
            for message in reversed(messageList.data):
                message_dict = {
                    "role": message.role,  # Replace with relevant attributes
                    "text": message.content[0].text.value,
                    # ...other attributes as needed
                } 
                messages.append(message_dict) 
            # print(messages)
            logging.info(f"Retrieved messages for thread_id {thread_id} and user_id {user_id}")
            emit('history_response', {"status": True, "messages": messages}, to=request.sid)
        else:
            logging.warning(f"Failed to find data for thread_id {thread_id} or user_id {user_id}")
            emit('error_response_history', {"status": False, "error": "No Thread or User"}, to=request.sid)
    except KeyError as e:
        # This block catches missing keys in the JSON request
        # Log and handle missing keys in the JSON request
        logging.error(f"Missing key in request data: {e.args[0]}")
        emit('error_response_history', {"status": False, "error": f"Missing key: {e.args[0]}"}, to=request.sid)
     
    except Exception as e:
        # This block catches any other exception, including errors from your custom modules
        logging.error(f"An unexpected error occurred: {str(e)}")
        emit('error_response_history', {"status": False, "error": str(e)}, to=request.sid)
    

 
@socketio.on('cancel_query')
def cancel_run(query):
    try: 
        session_id = query.get('sessionId')  
        print("in socket",session_id)
        logging.info(f"IN Run {session_id} 143")
        matched_item = None
        for item in active_thread.thread:
            if item['user_identifier'] == session_id:
                matched_item = item
        if matched_item is not None:
            acknowledge = openai_fun.cancel_run(session_id)
            print(traceback.format_exc())
            logging.info(f"Run Cancelled") 
            # emit('query_response', {"status": True, "thread_id": "thread_GYIhhy3uih2yYGzs6LSuK0q1", "end":"!@#$%", "response": "Something Went Wrong"}, to=session_id)
            print("//done//") 
            gevent.sleep(1)  #added
            emit('query_response', {"status": True, "thread_id": matched_item["thread_id"], "end":"!@#$%", "response": "@//done//@"}, to=session_id)
            emit('cancel_query_response',{"status":True, "acknowlwdge":acknowledge}, to=session_id)
            active_thread.thread.remove(matched_item)
        else:
            emit('cancel_query_response',{"status":True, "acknowlwdge":f"No run for {session_id}"}, to=session_id)
    except Exception as e:
        logging.error(f"Error massege : {e.args[0]}")
        emit('error_response_cancel', {"status": False, "error": f"Failed to cancel run, {str(e)}"}, to=session_id)
  

if __name__ == '__main__':
    socketio.run(app, port=2000, debug=True)
    # socketio.run(app, host='0.0.0.0', port=2000, debug=False)

server screen shot

I know its little complex but please help me
@lucaswiman @dbrnz @spezifanta @miguelgrinberg please look into this

[miguelgrinberg]

Try this:

• uninstall eventlet, gevent, gevent-websocket and anything else related to eventlet or gevent
• remove eventlet monkey patching and any references to eventlet in your application
• if you were explicitly setting async_mode='eventlet', remove that and let Socket.IO pick the default async mode for you
• if you are using gunicorn on your production server, remove -k eventlet, and add --threads 100 instead. You can change the 100 to a different number. This is the maximum number of connections the server will accept.

Try the application after making these changes and let me know if you see any difference.

[kashish63]

okay @miguelgrinberg then what about making so many connections. Please help me understand why that is happening. I am using websocket transport mode on the client side, so why is it disconnecting again and again?

[miguelgrinberg]

Did you look at the server logs? If the connections are failing the logs should show this.

[kashish63]

Yes, I saw the logs; they only show connection successful and then disconnect. What could be a problem?

Is there anything wrong with my code or something that I am missing?
I want to know one thing: Is it possible to make a connection that will not disconnect until the user leaves or refreshes the page? Currently, when I chat with my chatbot, it makes a new connection in between sending replies, and I am sending a response to that previous connection, through which I got a question.
@miguelgrinberg Please help me in this

[miguelgrinberg]

Did you implement a single-page app, or are you navigating between pages in this app? For the connection to not break you have to stay in the same page, the browser ends all websockets when it navigates to a new page.

[kashish63]

Its a single page application.
currently I am using eventlet again and it now it is not making connection again and again using socketio = SocketIO(app,cors_allowed_origins="*", async_mode='eventlet', ping_interval=10, ping_timeout=60)

running application using: gunicorn -b 0.0.0.0:2000 --worker-class eventlet -w 1 app_chatbot:app
but when disconnect of socket it shows error
socket shutdown error: [Errno 9] Bad file descriptorI NFO:root:Disconnected: noxyoDxPnHshsBB2AAAD
is that a major issue?

[kashish63]

hai @miguelgrinberg
I am writing code from starch for python socket io with acyncio. In my code, I am calling an event. That event contains a for loop and an emitting. but in the meantime, I called another event that previous one didn't complete. So how can I stop that event immediately?
Please suggest some ideas for that use case.

[miguelgrinberg]

Can you show me the code that you wrote? I don't fully understand what problem you are trying to solve. Socket.IO works asynchronously, so you do not need to stop an event to send another one.

[kashish63]

First of all, I want to thank you so much for your efforts and for your concise response. Here is my code,
first the starting of my code which has imports and connect disconnect function.

import socketio
from datetime import datetime    
import logging 
import traceback 
from utils import openai_fun, functions, active_thread
from database import mongo_database
# create a Socket.IO server
sio = socketio.AsyncServer(cors_allowed_origins="*", async_mode="asgi")   #engineio_logger=True logger=True
  
# wrap with ASGI application
app = socketio.ASGIApp(sio, static_files = {
    '/': './public/',   
})  

 
@sio.event 
async def connect(sid, environ):
    print(sid, 'Connected')
 

@sio.event
async def disconnect(sid):
    print(sid, 'Disconnected')

It only has two events in it.

1. gpt_query 2. cancel_query
   1. gpt_query : when this event calls, it takes question in this payload as request then process with GPT_Query2 function, that GPT_Query2 function takes sid, sio which is session ID and socket through which i am emitting response to user in GPT_Query2 function. So GPT_Query2 will generate response for user question and emit to it. It emit word by word by which this function complete execution in 5-10 seconds and continuously sending data to user .

@sio.on('gpt_query')
async def handle_gpt_query(sid, data):
    print("data is ", data)
    try:
        question = data.get('question', '').strip()
        thread_id = data.get('thread_id')
        user_id = data.get('user_id', '').strip()
        if not user_id:
            await sio.emit('error_response', {"status": False, "error": "User ID is missing or empty"}, to=sid)
            return 
        if not question:  
            await sio.emit('error_response', {"status": False, "error": "Question is missing or empty"}, to=sid)
            return
        if not thread_id: 
            thread = openai_fun.create_thread()
            thread_data = {  
                "user_id": user_id,
                "thread_id": thread.id,
                "query_head": question[:40],
                "created_at": datetime.now(), 
                "isDeleted": False, 
            }   
            mongo_database.insert_data(thread_data)  
            logging.info(f"New thread created with ID: {thread_id} for user: {user_id}")

        await openai_fun.GPT_Query2(thread_id, question, sid, sio, credentials) 

    except KeyError as e: 
        logging.error(f"Missing key: {e.args[0]}")
        await sio.emit('error_response', {"status": False, "error": f"Missing key: {e.args[0]}"}, to=sid)

    except Exception as e: 
        logging.error(f"Unhandled exception: {str(e)}")
        await sio.emit('error_response', {"status": False, "error": str(e)}, to=sid)

2. cancel_query: This is my another event that will call while gpt_query event emitting data to user. This event will cancel run of gpt response and emit @//done//@ to user after canceling response. But this is not emitting @//done//@ to user after canceling instead storing some data in its buffer. To resolve this condition I have to use await sio.sleep(1). this is clearing the buffer and emit @//done//@ to user.

@sio.on('cancel_query')
async def cancel_run(sid, query):
    try: 
        session_id = query.get('sessionId')    
        print("in socket",session_id)
        matched_item = next((item for item in active_thread.thread if item['user_identifier'] == session_id), None)
        if matched_item is not None:
            acknowledge = await openai_fun.cancel_run(session_id)
            logging.info(f"Run Cancelled") 
            await sio.sleep(1) 
            await sio.emit('query_response', {"status": True, "thread_id": matched_item["thread_id"]"end: "end":"!@#$%", "response": "@//done//@"}, to=session_id)
            await sio.emit('cancel_query_response',{"status":True, "acknowlwdge":acknowledge}, to=sid)
            active_thread.thread.remove(matched_item)
        else:
            await sio.emit('cancel_query_response',{"status":True, "acknowlwdge":f"No run for {session_id}"}, to=session_id)
    except Exception as e:
        logging.error(f"Error massege : {e.args[0]}")
        await sio.emit('error_response_cancel', {"status": False, "error": f"Failed to cancel run, {str(e)}"}, to=session_id)    

I am running this code using uvicorn app:app
Issue:

1. that await sio.sleep(1) takes time to emit final response to user.
2. Is that code good for deployment if not then please tell me the right approach using oops concept or anything else.
3. I am canceling run from open ai that process take 1-2 second, so can i just stop gpt_query for emitting data when user hit cancel_query.
4. will ever this code create conflict in future as eventlet did.
5. I am running this code with out if name == "main": and its running fine on my local computer. Should I use it in production level code.

I know some of my question being silly. I am deploy my app for the first time and I am bit confused.

Thanks again :)

[miguelgrinberg]

What is openai_fun? That is important, it is unclear to me if this is a library that works with threads or with asyncio.

[kashish63]

no that is a python file called openai_fun.py in which i define GPT_Query2 function. that function calls open ai gpt 4 model and then emits gpt 4 response to user; its like a chatbot. this GPT_Query2 function takes 5–10 seconds to complete its execution

[miguelgrinberg]

Unfortunately there is nothing I can see that is specific to Socket.IO that is wrong. My guess is that the way you manage the cancellation of a query is causing this.

[kashish63]

I just want to know if there is any way to stop emitting words to the user when the cancel button is clicked. Can I do this with create_task and then cancel() that task? Can I deploy this pattern code?

[miguelgrinberg]

You can use any mechanism offered by Python for this. I wrote a blog post long ago with some ideas on how to do this. You will need to adapt it to asyncio, but the general ideas are the same and you should be able to use an Event object for this.