分布式任务调整

logic/logic_main_service.py

@@ -22,29 +22,30 @@ def get_task_distributed():
             result = main_service_db.db_task_distributed("status = 1 order by priority desc,created_at asc limit 1 for update")
             if result:
                 #获取需要执行的步骤
-                next_step = need_run_stepx(result["id"])
-                if next_step == "no" or next_step == "error":
-                    print("获取需要执行的步骤 next_step",next_step)
-                    return next_step
+                #next_step = need_run_stepx(result["id"])
+                next_step_result = need_run_step_no_step1()
+                if next_step_result == "no" or next_step_result == "error":
+                    print("获取需要执行的步骤 next_step",next_step_result)
+                    return next_step_result
                 #非R11 R12 的主机在执行step2的时候,需要判断当前模型是否需要高精模或者photo3参与建模，如果是的话，该主机不执行这一步
-                if next_step == "step2":
-                    if common.task_need_high_model_or_photo3(result["task_key"]):
-                        print(f'模型{result["task_key"]}需要高精模或者photo3参与建模,该主机{hostname}不执行step2')
+                if next_step_result["run_step"] == "step2":
+                    if common.task_need_high_model_or_photo3(next_step_result["task_key"]):
+                        print(f'模型{next_step_result["task_key"]}需要高精模或者photo3参与建模,该主机{hostname}不执行step2')
                         return "no"
-                taskData = {"hostname":hostname,"run_step":next_step,"task_distributed_id":result["id"],"task_key":result["task_key"]}
-                flagRes = update_main_and_add_detail(taskData)
+                #taskData = next_step_result #{"hostname":hostname,"run_step":next_step,"task_distributed_id":result["id"],"task_key":result["task_key"]}
+                flagRes = update_main_and_add_detail(next_step_result)
                 if flagRes == "error":
                     print(f'出现错误,有可能是多个进程获取同一个任务了')
                     return "error"
-                print(f'任务ID-{taskData["task_key"]}- "执行{taskData["run_step"]}"')
-                return taskData
+                print(f'任务ID-{next_step_result["task_key"]}- "执行{next_step_result["run_step"]}"')
+                return next_step_result
             else:
                 return 'no_data'
         else:
             #R11 R12的主机 可以执行step1 2 3 的任务
             #如果R11 R12的主机目前没有正在执行step2,则优先处理step2,
             # print("次数",is_run_stepx_nums("step2"))
-            if is_run_stepx_nums("step2") < 2:
+            if is_run_stepx_nums("step2") < 0:
                 resultData = need_run_step2()
                 if resultData != "no": 
                     resultData["hostname"] = hostname
@@ -187,6 +188,27 @@ def need_run_step_no_step2():
         print(f"{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())} 执行need_run_step_no_step2()异常: {str(e)}")
         logging.error(f"{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())} 执行need_run_step_no_step2()异常: {str(e)}")
         return 'error'
+
+#查询出哪些任务需要执行非step1
+def need_run_step_no_step1():
+    try:
+        result = main_service_db.db_task_distributed_list("finished_at is null order by priority desc,created_at asc for update")
+        #判断是否有值
+        if len(result) == 0:
+            return "no"
+        #遍历循环哪笔需要执行step2
+        for row in result:
+            #判断是否有正在执行的step2
+            xstep = need_run_stepx(row["id"])
+            #print("查询非step1的任务列表",xstep,row["id"])
+            if xstep == "step2" or xstep == "step3":
+                #没有正在执行的step1,则返回该任务
+                return {"hostname":hostname,"run_step":xstep,"task_distributed_id":row["id"],"task_key":row["task_key"]}
+        return "no"
+    except Exception as e:
+        print(f"{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())} 执行need_run_step_no_step1()异常: {str(e)}")
+        logging.error(f"{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())} 执行need_run_step_no_step1()异常: {str(e)}")
+        return 'error'
             
 #更新主表和插入明细表的步骤
 def update_main_and_add_detail(data):