[Python] Why am I receiving an Errno 98: Address already in use?

Ok, so I have 3 computers. 1 runs windows 11, and 2 run Ubuntu. The Windows machine is my test machine, so it runs my test programs to get out the bugs. The other 2 machines are my servers. One runs a MySQL and FTP server, and the other runs a Queue server that has 5 workers for it. When I run the test program, it makes the connection to the Queue server, and sends the command. The Queue server successfully parses the command and finds the correct program to call. But every time calls a program, I receive the:

OSError: [Errno 98] Address already in use

I cannot figure out what is causing the problem.

But here is some of the Code. I can upload more if it helps for different questions. As far as I can tell from the callbacks, it doesn't actually get to calling the program.

Test Program:

# 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 Program:

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"))


Sample Callback from Test Program:

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 <module>
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


I have managed to try a few things. I have tried making sure the ports are freed up and allowed by the firewall, on both the Queue Server and the MySQL Server going both ways. So both ports are available on each machine. I have double checked that the IP Address for the MySQL server is correct. I have even double checked to make sure that the Port is available to reuse after a disconnect and restart:

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)


I have even shorted the wait time in the system settings for the ip/port address. I can ping the MySQL Machine all day long from the Queue machine and receive anywhere from 1 to 28 ms in response time. I have triple checked as to make sure the commands sent are all formatted the way they are needed to interact with the server. But I still end up with the traceback above. I have also checked if the port is being used by any application using the lsof -i :8888. Before I start the Queue server, I show nothing. When I start it, just the Queue server is showing listening to the port. And I restart the Queue machine before I start the Queue server to make sure the instances are cleaned up.

So line 84 is the client_socket.close(). But here is the entire run function:

def run(self):
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)

self.server_running = True
while self.server_running:
try:
client_socket, client_ip = server_socket.accept()
self.log_function(f"Connection from {client_ip}")

if client_ip[0] not in self.ALLOWED_IPs:
self.log_function(f"Rejected connection from {client_ip}")
client_socket.sendall(b"Error: Connection Not Allowed")
client_socket.close()
continue

data = client_socket.recv(1024).decode("utf-8").strip()
if not data:
self.log_function(b"Error: decoding data {e}.")
client_socket.sendall(b"Error: Invalid Data Format.")
client_socket.close()
continue

try:
username, command, contents = data.split("|")
except ValueError:
client_socket.sendall(b"Error: Invalid request format.")
client_socket.close()
continue

with queue_lock:
command_queue.put((username, command, contents, client_socket))
except Exception as e:
self.log_function(f"Error: {str(e)}")
server_socket.close()

Continue reading...