FIX AutoDL修复临界区资源竞争问题

AutoDL/autodl_scheduling/entity/instance_pool.py

@@ -74,51 +74,65 @@ class InstancePool:
 
     def introspection(self):
         # 停止超时实例(运行超时和无任务超时)
-        before=len(self.instances)
+        timeout = self.timeout//6
-        instance_copy = copy.deepcopy(self.instances)
+        while len(self.intro_threads) > 0 and timeout > 0:
-        flag = False
+            time.sleep(1)
-        for instance in instance_copy:
+            timeout -= 1
-            if instance.active:
+        if timeout == 0:
-                if (time.time() - instance.last_active_time) > self.timeout:
+            loguru.logger.error("Fail to Introspect Instances: Timeout")
-                    flag = True
+            return
-                    self.intro_threads.append(self.executor.submit(self.remove_instance, instance=instance))
+        try:
-            else:
+            before=len(self.instances)
-                if (time.time() - instance.last_active_time) > self.scaledown_window:
+            instance_copy = copy.deepcopy(self.instances)
-                    flag = True
+            flag = False
-                    self.intro_threads.append(self.executor.submit(self.remove_instance, instance=instance))
+            for instance in instance_copy:
-        while len(self.intro_threads) > 0:
+                if instance.active:
-            for t in self.intro_threads:
+                    if (time.time() - instance.last_active_time) > self.timeout:
-                t.result(timeout=self.timeout//2)
+                        flag = True
-                self.intro_threads.remove(t)
+                        self.intro_threads.append(self.executor.submit(self.remove_instance, instance=instance))
-        after = len(self.instances)
+                else:
-        if flag:
+                    if (time.time() - instance.last_active_time) > self.scaledown_window:
-            loguru.logger.info("Instance Num Before Introspecting %d After Introspecting %d" % (before, after))
+                        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()
+        try:
-        # 调整实例数量
+            self.introspection()
-        instance_copy = copy.deepcopy(self.instances)
+            # 调整实例数量
-        dest = target_instance - len(instance_copy)
+            instance_copy = copy.deepcopy(self.instances)
-        if (disable_shrink and dest < 0) or dest == 0:
+            dest = target_instance - len(instance_copy)
-            return True
+            if (disable_shrink and dest < 0) or dest == 0:
-        loguru.logger.info("Instance Num Before Scaling %d ; Target %d" % (len(self.instances), target_instance))
+                return True
-        if dest < 0:
+            loguru.logger.info("Instance Num Before Scaling %d ; Target %d" % (len(self.instances), target_instance))
-            dest = abs(dest)
+            if dest < 0:
-            for instance in instance_copy:
+                dest = abs(dest)
-                if not instance.active and dest > 0:
+                for instance in instance_copy:
-                    self.threads.append(self.executor.submit(self.remove_instance, instance=instance))
+                    if not instance.active and dest > 0:
-                    dest -= 1
+                        self.threads.append(self.executor.submit(self.remove_instance, instance=instance))
-        elif dest > 0:
+                        dest -= 1
-            for i in range(dest):
+            elif dest > 0:
-                self.threads.append(self.executor.submit(self._add_instance))
+                for i in range(dest):
-        while len(self.threads) > 0:
+                    self.threads.append(self.executor.submit(self._add_instance))
-            for t in self.threads:
+            while len(self.threads) > 0:
-                t.result(timeout=self.timeout//2)
+                for t in self.threads:
-                self.threads.remove(t)
+                    t.result(timeout=self.timeout//2)
-        loguru.logger.info("Instance Num After Scaling %d ; Target %d" % (len(self.instances), target_instance))
+                    self.threads.remove(t)
-        if len(self.instances) == target_instance:
+            loguru.logger.info("Instance Num After Scaling %d ; Target %d" % (len(self.instances), target_instance))
-            return True
+            if len(self.instances) == target_instance:
-        else:
+                return True
+            else:
+                return False
+        except Exception as e:
+            loguru.logger.error("Fail to Scale: %s" % e)
             return False
 
 

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

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)