proxy pool race condition fixed

freeplay2api.py

@@ -343,45 +343,49 @@ class KeyMaintainer(threading.Thread):
         while True:
             try:
                 logging.info("KeyMaintainer: Starting maintenance cycle.")
-                
-                # First, update balances of existing accounts
                 accounts = self.manager.get_all_accounts()
+
+                # 更新所有账户余额
                 for account in accounts:
                     balance = self.client.check_balance(account["session_id"])
                     if balance != account.get("balance"):
                         account["balance"] = balance
                         self.manager.update_account(account)
-                        logging.info(f"Account {account['email']} balance updated to ${balance:.4f}")
-
-                # Now, check if we need to register new accounts and loop until we have enough
-                while True:
-                    healthy_accounts = [
-                        acc for acc in self.manager.get_all_accounts()
-                        if acc.get("balance", 0) > self.config["LOW_BALANCE_THRESHOLD"]
-                    ]
-                    needed = self.config["ACTIVE_KEY_THRESHOLD"] - len(healthy_accounts)
-
-                    if needed <= 0:
-                        logging.info(f"Sufficient healthy accounts ({len(healthy_accounts)}). No registration needed for now.")
-                        break
-
-                    logging.info(f"Healthy accounts ({len(healthy_accounts)}) below threshold ({self.config['ACTIVE_KEY_THRESHOLD']}). Need to register {needed} new accounts.")
-                    
-                    batch_size = min(needed, self.config["REGISTRATION_CONCURRENCY"])
-                    with concurrent.futures.ThreadPoolExecutor(max_workers=batch_size) as executor:
-                        futures = [executor.submit(self.client.register) for _ in range(batch_size)]
-                        
-                        # Wait for the batch to complete
-                        concurrent.futures.wait(futures)
-
-                    logging.info("Registration batch finished. Re-checking healthy account count.")
-                    # The loop will now re-evaluate the 'needed' count
+                        logging.info(
+                            f"Account {account['email']} balance updated to ${balance:.4f}"
+                        )
+
+                # 检查是否需要注册新账号
+                healthy_accounts = [
+                    acc
+                    for acc in self.manager.get_all_accounts()
+                    if acc.get("balance", 0) > self.config["LOW_BALANCE_THRESHOLD"]
+                ]
+                needed = self.config["ACTIVE_KEY_THRESHOLD"] - len(healthy_accounts)
+
+                if needed > 0:
+                    logging.info(
+                        f"Healthy accounts ({len(healthy_accounts)}) below threshold ({self.config['ACTIVE_KEY_THRESHOLD']}). Need to register {needed} new accounts."
+                    )
+                    with concurrent.futures.ThreadPoolExecutor(
+                        max_workers=self.config["REGISTRATION_CONCURRENCY"]
+                    ) as executor:
+                        futures = [
+                            executor.submit(self.client.register) for _ in range(needed)
+                        ]
+                        for future in concurrent.futures.as_completed(futures):
+                            new_account = future.result()
+                            if new_account:
+                                self.manager.add_account(new_account)
+                else:
+                    logging.info(
+                        f"Sufficient healthy accounts ({len(healthy_accounts)}). No registration needed."
+                    )
 
             except Exception as e:
                 logging.error(f"Error in KeyMaintainer cycle: {e}")
 
-            logging.info(f"KeyMaintainer cycle finished. Waiting for {self.config['CHECK_INTERVAL_SECONDS']} seconds.")
-            time.sleep(self.config['CHECK_INTERVAL_SECONDS'])
+            time.sleep(self.config["CHECK_INTERVAL_SECONDS"])
 
 
 # --- FastAPI应用 ---
@@ -435,8 +439,18 @@ def initialize_app():
             logging.info("Initializing proxy pool...")
             proxy_pool_config = config.get("PROXY_POOL_CONFIG", {})
             proxy_pool = ProxyPool(proxy_pool_config)
-            proxy_pool.initialize()
-            logging.info("Proxy pool initialized.")
+            
+            # Start initialization in a separate thread
+            init_thread = threading.Thread(target=proxy_pool.initialize, daemon=True)
+            init_thread.start()
+            
+            # Wait for the pool to be initialized
+            logging.info("Waiting for proxy pool to complete initial refill...")
+            while not proxy_pool.is_initialized:
+                time.sleep(1)
+            init_thread.join() # Ensure initialization thread is complete
+            logging.info("Proxy pool initialization process finished.")
+
         else:
             logging.info("Proxy pool is disabled in config.")
 
@@ -446,7 +460,6 @@ def initialize_app():
 
         # 5. 启动后台维护线程
         maintainer = KeyMaintainer(account_manager, freeplay_client, config)
-        # Always start the maintainer, it will wait for the pool internally if needed.
         maintainer.start()
         logging.info("Key maintenance service started.")