From 1dfab45d83cef04f112d0af9f46229190580c3be Mon Sep 17 00:00:00 2001 From: "kyj@bowong.ai" Date: Thu, 17 Apr 2025 15:37:31 +0800 Subject: [PATCH] =?UTF-8?q?FIX=20AutoDL=E4=BF=AE=E5=A4=8D=E4=B8=B4?= =?UTF-8?q?=E7=95=8C=E5=8C=BA=E8=B5=84=E6=BA=90=E7=AB=9E=E4=BA=89=E9=97=AE?= =?UTF-8?q?=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../autodl_scheduling/entity/instance_pool.py | 100 ++++++++++-------- .../autodl_scheduling/entity/running_pool.py | 1 + AutoDL/autodl_scheduling/server.py | 18 +++- 3 files changed, 72 insertions(+), 47 deletions(-) diff --git a/AutoDL/autodl_scheduling/entity/instance_pool.py b/AutoDL/autodl_scheduling/entity/instance_pool.py index 947a1b6..e4554c0 100644 --- a/AutoDL/autodl_scheduling/entity/instance_pool.py +++ b/AutoDL/autodl_scheduling/entity/instance_pool.py @@ -74,51 +74,65 @@ class InstancePool: def introspection(self): # 停止超时实例(运行超时和无任务超时) - before=len(self.instances) - instance_copy = copy.deepcopy(self.instances) - flag = False - for instance in instance_copy: - if instance.active: - if (time.time() - instance.last_active_time) > self.timeout: - flag = True - self.intro_threads.append(self.executor.submit(self.remove_instance, instance=instance)) - else: - if (time.time() - instance.last_active_time) > self.scaledown_window: - flag = True - self.intro_threads.append(self.executor.submit(self.remove_instance, instance=instance)) - while len(self.intro_threads) > 0: - for t in self.intro_threads: - t.result(timeout=self.timeout//2) - self.intro_threads.remove(t) - after = len(self.instances) - if flag: - loguru.logger.info("Instance Num Before Introspecting %d After Introspecting %d" % (before, after)) + timeout = self.timeout//6 + while len(self.intro_threads) > 0 and timeout > 0: + time.sleep(1) + timeout -= 1 + if timeout == 0: + loguru.logger.error("Fail to Introspect Instances: Timeout") + return + try: + before=len(self.instances) + instance_copy = copy.deepcopy(self.instances) + flag = False + for instance in instance_copy: + if instance.active: + if (time.time() - instance.last_active_time) > self.timeout: + flag = True + self.intro_threads.append(self.executor.submit(self.remove_instance, instance=instance)) + else: + if (time.time() - instance.last_active_time) > self.scaledown_window: + flag = True + self.intro_threads.append(self.executor.submit(self.remove_instance, instance=instance)) + while len(self.intro_threads) > 0: + for t in self.intro_threads: + t.result(timeout=self.timeout//2) + self.intro_threads.remove(t) + after = len(self.instances) + if flag: + loguru.logger.info("Instance Num Before Introspecting %d After Introspecting %d" % (before, after)) + except Exception as e: + loguru.logger.error("Fail to Introspect Instances: %s" % e) def _scale(self, target_instance:int, disable_shrink=True): - self.introspection() - # 调整实例数量 - instance_copy = copy.deepcopy(self.instances) - dest = target_instance - len(instance_copy) - if (disable_shrink and dest < 0) or dest == 0: - return True - loguru.logger.info("Instance Num Before Scaling %d ; Target %d" % (len(self.instances), target_instance)) - if dest < 0: - dest = abs(dest) - for instance in instance_copy: - if not instance.active and dest > 0: - self.threads.append(self.executor.submit(self.remove_instance, instance=instance)) - dest -= 1 - elif dest > 0: - for i in range(dest): - self.threads.append(self.executor.submit(self._add_instance)) - while len(self.threads) > 0: - for t in self.threads: - t.result(timeout=self.timeout//2) - self.threads.remove(t) - loguru.logger.info("Instance Num After Scaling %d ; Target %d" % (len(self.instances), target_instance)) - if len(self.instances) == target_instance: - return True - else: + try: + self.introspection() + # 调整实例数量 + instance_copy = copy.deepcopy(self.instances) + dest = target_instance - len(instance_copy) + if (disable_shrink and dest < 0) or dest == 0: + return True + loguru.logger.info("Instance Num Before Scaling %d ; Target %d" % (len(self.instances), target_instance)) + if dest < 0: + dest = abs(dest) + for instance in instance_copy: + if not instance.active and dest > 0: + self.threads.append(self.executor.submit(self.remove_instance, instance=instance)) + dest -= 1 + elif dest > 0: + for i in range(dest): + self.threads.append(self.executor.submit(self._add_instance)) + while len(self.threads) > 0: + for t in self.threads: + t.result(timeout=self.timeout//2) + self.threads.remove(t) + loguru.logger.info("Instance Num After Scaling %d ; Target %d" % (len(self.instances), target_instance)) + if len(self.instances) == target_instance: + return True + else: + return False + except Exception as e: + loguru.logger.error("Fail to Scale: %s" % e) return False diff --git a/AutoDL/autodl_scheduling/entity/running_pool.py b/AutoDL/autodl_scheduling/entity/running_pool.py index 36aa859..27a9350 100644 --- a/AutoDL/autodl_scheduling/entity/running_pool.py +++ b/AutoDL/autodl_scheduling/entity/running_pool.py @@ -59,6 +59,7 @@ class RunningPool: "audio_file": audio_file } resp = requests.post(base_url, data, headers=self._headers, allow_redirects=True, stream=True) + loguru.logger.info("Submit Response: " % resp.text) if resp.status_code == 200: if resp.json()["status"] == "success": code = resp.json()["code"] diff --git a/AutoDL/autodl_scheduling/server.py b/AutoDL/autodl_scheduling/server.py index 1d79ccf..f7458b4 100644 --- a/AutoDL/autodl_scheduling/server.py +++ b/AutoDL/autodl_scheduling/server.py @@ -21,13 +21,15 @@ class Server: self.app = FastAPI() self.waiting_queue = WaitingQueue() self.running_pool = RunningPool() - self.instance_pool = InstancePool(max_instance=29) + #账号限制max_instance不能超过30 + self.instance_pool = InstancePool(max_instance=5) self.result_map = ResultMap() self.executor = ThreadPoolExecutor(max_workers=2) self.worker_1 = self.executor.submit(self.scaling_worker) self.worker_2 = self.executor.submit(self.introspect_instance) self.app.routes.append(APIRoute("/", endpoint=self.submit, methods=['POST'])) self.app.routes.append(APIRoute("/{uid}", endpoint=self.get_result, methods=['GET'])) + self.app.routes.append(APIRoute("/", endpoint=self.get_all_result, methods=['GET'])) # async def submit(self, video_url, audio_url, token: HTTPAuthorizationCredentials = Depends(HTTPBearer())): async def submit(self, video_url, audio_url): @@ -48,10 +50,14 @@ class Server: try: return self.result_map.get(uid) except: - if uid in self.running_pool.tasks: - return {"status": "running", "msg": ""} return {"status": "queuing", "msg":""} + async def get_all_result(self): + try: + return self.result_map.map + except: + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR) + def introspect_instance(self): loguru.logger.info("Introspecting worker started || Scaledown Window: %ds" % self.instance_pool.scaledown_window) while True: @@ -74,23 +80,27 @@ class Server: instance.active = True instance.last_active_time = time.time() loguru.logger.info("Task Submitted") + self.result_map.set(uid, {"status": "running", "msg": "Task Submitted, Waiting for result"}) else: loguru.logger.error("Submit Task Failed") + self.result_map.set(uid, {"status":"fail", "msg":"Submit Task Failed"}) #查询结果放到结果缓存里 for uid, task in list(self.running_pool.tasks.items()): result = self.running_pool.result(uid) # 任务运行完成或失败 if result: - self.result_map.set(uid, result) + self.result_map.update(uid, result) self.running_pool.tasks.pop(uid) if result["status"] == "fail": #删除失败任务的实例 self.instance_pool.remove_instance(Instance(uuid=task["instance_id"])) + loguru.logger.info("Instance[%s] Removed Due to Task Failure" % task["instance_id"]) else: #更新成功任务实例的状态 for instance in self.instance_pool.instances: if instance.uuid == task["instance_id"]: + loguru.logger.info("Instance[%s] Task Finished" % instance.uuid) instance.active = False instance.last_active_time = time.time() time.sleep(0.5)