奇怪的python异步行为

Question

上下文：我想查询内存中的 dict 结构并使用相同的查询函数来触发一些后处理任务（例如获取附加到 dict 的扩展信息，因此名称为 exinfo）因为我不想让那些在“关键处理路径”中。 Asyncio 表现得很奇怪，可能是因为我在这里做错了一些事情。 task_resolve_names 和 task_save 不会被执行，除非我在运行 run_forever 线程之前取消注释“hello world”测试任务。感谢您的帮助。

exinfo.py

import os, sys
from os import path
import pickle
import asyncio
import threading
import sys
print (sys.version)
print (sys.version_info)

exinfopath = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data")
extpkl = os.path.join(exinfopath, "exinfo.pkl")

exinfo = None

class ExInfo:
    instance = None
    
    def __new__(cls, exinfopath, extpkl):
        if not ExInfo.instance:
            ExInfo.instance = ExInfo.__ExInfo(exinfopath, extpkl)
        return ExInfo.instance

    class __ExInfo:
        client = None
        ext = None
        count = 0
        lock = asyncio.Lock()
        loop = None
        extpkl = None
        def __init__(self, exinfopath, extpkl): #arg TBD
            self.loop = asyncio.get_event_loop()
            #self.loop.create_task(self.hello()) <--- weird behavior, UNCOMMENT this to mk stuff happen
            threading.Thread(target=self.thread_main_loop, 
                args=[self.lock, self.loop],
                daemon=True).start()

        def query(self, host, dnsquery): #host can be ipaddr or not
            r = {}
            r['hits'] = 1
            r['query'] = dnsquery
            print("query")
            
            if 'post_process1' not in r.keys():
                print("self.loop.create_task")
                self.loop.create_task(self.task_resolve_names(r, self.lock))
            
            if self.count % 5 == 0: #save pickle every 5 executions
                print("self.loop.create_task")
                self.loop.create_task(self.task_save(r, self.lock))
            
            return r
        
        def thread_main_loop(self, lock, loop):
            print("thread_main_loop / loop.run_forever()")
            loop.run_forever()
            loop.close()

        async def hello(self):
            print("WORLD")

        async def task_resolve_names(self, ext, lock):
            print("task_resolve_names")
            async with lock:
              ext['post_process1'] = 'OK'
              
        
        async def task_save(self, ext, lock):
             print("task_save")
             async with lock:
                ext['post_process2'] = 'OK'
                #with open(self.extpkl, 'wb') as f:
                #    pickle.dump(self.ext, f, protocol=pickle.HIGHEST_PROTOCOL)

def init():
    global exinfo
    exinfo = ExInfo(exinfopath, extpkl)

def test_query(info, query):
    global exinfo
    print("{}".format(exinfo.query(info, query)))

test_exinfo.py

#!/usr/bin/env python3
import os
import exinfo
import time
exinfopath = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data")
extpkl = os.path.join(exinfopath, "exinfo.pkl")

exinfo.init()
exinfo.test_query("someinfo", "someinfo")

time.sleep(10)

实际结果：

3.7.6 (default, Jan  8 2020, 13:42:34) 
[Clang 4.0.1 (tags/RELEASE_401/final)]
sys.version_info(major=3, minor=7, micro=6, releaselevel='final', serial=0)
thread_main_loop / loop.run_forever()
query
self.loop.create_task
self.loop.create_task
{'hits': 1, 'query': ''}

预期结果：

3.7.6 (default, Jan  8 2020, 13:42:34) 
[Clang 4.0.1 (tags/RELEASE_401/final)]
sys.version_info(major=3, minor=7, micro=6, releaselevel='final', serial=0)
thread_main_loop / loop.run_forever()
query
self.loop.create_task
self.loop.create_task
{'hits': 1, 'query': ''}
task_resolve_names
task_save

Answer 1

query 方法使用loop.create_task 将任务添加到在不同线程中运行但没有正确同步的事件循环中。要解决此问题，请改用asyncio.run_coroutine_threadsafe。