Почему я получаю сообщение Errno 98: адрес уже используется? ⇐ MySql

[Anonymous]

Хорошо, у меня 3 компьютера. 1 запускает Windows 11, а 2 — Ubuntu. Машина с Windows — моя тестовая машина, поэтому на ней запускаются мои тестовые программы для устранения ошибок. Остальные 2 машины — мои серверы. На одном работает MySQL и FTP-сервер, а на другом — сервер очереди, на котором работают 5 рабочих. Когда я запускаю тестовую программу, она устанавливает соединение с сервером очереди и отправляет команду. Сервер очереди успешно анализирует команду и находит нужную программу для вызова. Но каждый раз при вызове программы я получаю:

Код: Выделить всё

OSError: [Errno 98] Address already in use

Я не могу понять, в чем причина проблемы.
Но вот часть кода. Я могу загрузить больше, если это поможет по разным вопросам. Насколько я могу судить по обратным вызовам, на самом деле дело не доходит до вызова программы.
Тестовая программа:

Код: Выделить всё

# Send a command to the Queue Program and receive a response
def send_to_queue(command):
try:
print("Attempting Socket Connection.")

# Establish socket connection
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.connect((QUEUE_SERVER["host"], QUEUE_SERVER["port"]))

# Ensure command ends with a newline or appropriate delimiter (adjust as needed)
formatted_command = f"{command}\n"
sock.sendall(formatted_command.encode('utf-8'))
print(f"Sending Command: {formatted_command.strip()}")

# Receive the response
response = sock.recv(4096).decode('utf-8')
print("Response received:", response)

# Check for server-specific errors
if "Error:" in response:
logging.warning(f"Queue Server responded with an error: {response}")
else:
logging.info(f"Command successfully processed.  Response: {response}")

# Optionally, process or write the response
write_response_to_document(response)

except Exception as e:
logging.error(f"Failed to communicate with Queue Program: {e}")
raise

Программа очереди:

Код: Выделить всё

QUEUE_PORT = 8888
MYSQL_CONFIG = {
"host": "xxx.xxx.x.xx",
"port": 3306,
"user": "Queue",
"password": "*****N@j%",
"database": "users"
}

COMMAND_EXECUTION_MAP = {
"CREATE_DATABASE": "Create_Database.py",
"DELETE_DATABASE": "Delete_Database.py",
"SHOW_DATABASE": "List_Database.py",
"CREATE_TABLE": "Create_Table.py",
"DELETE_TABLE": "Delete_Table.py",
"LIST_TABLES": "List_Tables.py",
"INSERT_ROW": "Insert_Row.py",
"RETRIEVE_ROW": "Retrieve_Row.py",
"DELETE_ROW": "Delete_Row.py",
"INSERT_ROWS": "Insert_Rows.py",
"RETRIEVE_ROWS": "Retrieve_Rows.py",
"DELETE_ROWS": "Delete_Rows.py",
"NEW_USER": "New_User.py",
}

Код: Выделить всё

    def process_request(self, username, command, contents, client_socket):
try:
self.log_function(f"[Worker {self.thread_id}] Starting process_request...")
self.log_function(f"[Worker {self.thread_id}] Parameters received - Username: {username}, Command: {command}, Contents: {contents}")

# Step 1: Handle NEW_USER command
if command == "NEW_USER":
# Check if the username already exists
new_contents = f"users, UserAccounts, {username}"
user_check_response = self.execute_script(
client_socket, "Queue", "RETRIEVE_ROW", new_contents
)
if user_check_response:
client_socket.sendall(b"Error: User already exists.")
return

# Check if the provided privilege code matches the role
username, password, role, time = user_check_response.split(',')
contents = f"users, UserAccounts, {role}"
privilege_code_response = self.execute_script(
client_socket, "Queue", "RETRIEVE_ROW", contents
)
if not privilege_code_response or privilege_code_response["code"] != contents.split()[1]:
client_socket.sendall(b"Error: Invalid privilege code.")
return

# Step 2: Bypass checks if Contents contains username
if username in contents:
self.execute_script(client_socket, username, command, contents)
return

# Step 3: Standard checks for other commands
# Retrieve user role
contents = f"users, UserAccounts, {username}"
user_role_response = self.execute_script(
client_socket, "Queue", "RETRIEVE_ROW", contents
)
if not user_role_response:
client_socket.sendall(b"Error: User not found.")
return
user_role = user_role_response["role"]

# Retrieve command role
contents = f"users, Commands, {command}"
command_role_response = self.execute_script(
client_socket, "Queue", "RETRIEVE_ROW", contents
)
if not command_role_response:
client_socket.sendall(b"Error: Command not found.")
return
command_role = int(command_role_response["role"])

# Retrieve user privilege number
contents = f"users, New_Privilege_Table, {user_role}"
privilege_number_response = self.execute_script(
client_socket, "Queue", "RETRIEVE_ROW", contents
)
if not privilege_number_response:
client_socket.sendall(b"Error:  Privilege data not found.")
return
privilege_number = int(privilege_number_response["number"])

# Validate privilege number against command role
if privilege_number < command_role:
client_socket.sendall(b"Error: Insufficient privileges for this command.")
return

# Step 4: Validate table role (if applicable)
table_role_response = self.execute_script(
client_socket, "Queue", "RETRIEVE_ROW", f"Database({contents.split(',')[0]}), Table(Index), SearchItem({contents.split(',')[1]})"
)
if table_role_response:
table_role = int(table_role_response["role"])
if privilege_number < table_role:
client_socket.sendall(b"Error: Insufficient privileges for this index.")
return

# Step 5: Execute the command
self.execute_script(client_socket, username, command, contents)

except Exception as e:
error_message = f"[Worker {self.thread_id}] Error processing request: {e}"
self.log_function(error_message)
client_socket.sendall(f"Error: {e}".encode("utf-8"))

finally:
self.log_function(f"[Worker {self.thread_id}] Finished processing request.")

def execute_script(self, client_socket, username, command, contents):
try:
# Validate base path and script path
self.log_function(f"Base path: {base_path}")
self.log_function(f"Command: {command}")
script_name = COMMAND_EXECUTION_MAP.get(command)
if not script_name:
raise ValueError(f"Command not found in Execution Map: {command}")

script_path = os.path.abspath(os.path.join(base_path, "PythonScripts", script_name))
self.log_function(f"Generated script path: {script_path}")

if not os.path.exists(script_path):
raise FileNotFoundError(f"Script not found: {script_path}")

# Log execution attempt
self.log_function(f"Executing script: {script_path} with arguments: {username}, {command}, {contents}")

# Execute script
result = subprocess.run(
[sys.executable, script_path, username, command, contents],
capture_output=True, text=True
)

# Handle result
if result.returncode == 0:
self.log_function(f"Script executed successfully. Output: {result.stdout.strip()}")
client_socket.sendall(result.stdout.encode("utf-8"))
else:
self.log_function(f"Script execution failed.  Error: {result.stderr.strip()}")
client_socket.sendall(result.stderr.encode("utf-8"))

except Exception as e:
error_message = f"Error executing script: {e}"
self.log_function(error_message)
client_socket.sendall(error_message.encode("utf-8"))

Пример обратного вызова из тестовой программы:

Код: Выделить всё

Listing Tables...
Attempting Socket Connection.
Sending Command: Fred|LIST_TABLES|Users
Response received: Traceback (most recent call last):
File "Queue.py", line 64, in run
OSError: [Errno 98] Address already in use

Inserting A Row...
Traceback (most recent call last):
File "D:\Workspace\Test.py", line 519, in 
main()
File "D:\Workspace\Test.py", line 469, in main
insert_row(username, database, table, document)
File "D:\Workspace\Test.py", line 81, in insert_row
upload_file_to_ftp(local_filepath, ftp_dir)
File "D:\Workspace\Test.py", line 342, in upload_file_to_ftp
ftp.connect(FTP_CREDENTIALS["host"], FTP_CREDENTIALS["port"])
File "D:\Program Files\Python\lib\ftplib.py", line 158, in connect
self.sock = socket.create_connection((self.host, self.port), self.timeout,
File "D:\Program Files\Python\lib\socket.py", line 845, in create_connection
raise err
File "D:\Program Files\Python\lib\socket.py", line 833, in create_connection
sock.connect(sa)
ConnectionRefusedError: [WinError 10061] No connection could be made because the target machine actively refused it

Мне удалось попробовать несколько вещей. Я попытался убедиться, что порты освобождены и разрешены брандмауэром как на сервере очереди, так и на сервере MySQL в обоих направлениях. Таким образом, оба порта доступны на каждой машине. Я дважды проверил правильность IP-адреса сервера MySQL. Я даже дважды проверил, доступен ли порт для повторного использования после отключения и перезапуска:

Код: Выделить всё

        self.log_function(f"Starting Queue server on port {QUEUE_PORT}...")
self.log_function("Attempting to bind to port 8888...")
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_socket.bind(("0.0.0.0", QUEUE_PORT))
server_socket.listen(10)

Я даже сократил время ожидания в системных настройках для адреса IP/порта.
Я могу целый день пинговать машину MySQL с машины очереди и получать откуда угодно Время ответа от 1 до 28 мс.
Я трижды проверил, все ли отправленные команды отформатированы так, как они необходимы для взаимодействия с сервером. Но я все равно получаю обратную трассировку, указанную выше.
Я также проверил, используется ли порт каким-либо приложением, используя lsof -i :8888. Прежде чем я запущу сервер очереди, я ничего не показываю. Когда я запускаю его, отображается только сервер очереди, прослушивающий порт. И я перезапускаю машину очереди перед запуском сервера очереди, чтобы убедиться, что экземпляры очищены.

Подробнее здесь: https://stackoverflow.com/questions/793 ... ady-in-use