Defer more asyncio Lock creation

Must be created after loop is running for python 3.9.

confluent_server/confluent/collective/manager.py

@@ -65,14 +65,18 @@ def verify_stub(store, misc):
 class ContextBool(object):
     def __init__(self):
         self.active = False
-        self.mylock = asyncio.Lock()
+        self.mylock = None
 
     async def __aenter__(self):
         self.active = True
+        if self.mylock is None:
+            self.mylock = asyncio.Lock()
         return await self.mylock.__aenter__()
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
         self.active = False
+        if self.mylock is None:
+            self.mylock = asyncio.Lock()
         return await self.mylock.__aexit__(exc_type, exc_val, exc_tb)
 
 connecting = ContextBool()
@@ -98,6 +102,8 @@ async def connect_to_leader(cert=None, name=None, leader=None, remote=None, isre
                  'subsystem': 'collective'})
         return False
     async with connecting:
+        if cfm._initlock is None:
+            cfm._initlock = asyncio.Lock()
         async with cfm._initlock:
             # remote is a socket...
             banner = await tlvdata.recv(remote)  # the banner

confluent_server/confluent/config/configmanager.py

@@ -98,10 +98,10 @@ import yaml
 _masterkey = None
 _masterintegritykey = None
 _dirtylock = threading.RLock()
-_leaderlock = asyncio.Lock()
+_leaderlock = None
 _synclock = threading.RLock()
-_rpclock = asyncio.Lock()
-_initlock = asyncio.Lock()
+_rpclock = None
+_initlock = None
 _followerlocks = {}
 _config_areas = ('nodegroups', 'nodes', 'usergroups', 'users')
 tracelog = None
@@ -472,6 +472,9 @@ def init_masterkey(password=None, autogen=True):
 
 
 async def _push_rpc(stream, payload):
+    global _rpclock
+    if _rpclock is None:
+        _rpclock = asyncio.Lock()
     async with _rpclock:
         try:
             stream[1].write(struct.pack('!Q', len(payload)))
@@ -861,6 +864,9 @@ class StreamHandler(object):
 
 
 async def stop_following(replacement=None):
+    global _leaderlock
+    if _leaderlock is None:
+        _leaderlock = asyncio.Lock()
     async with _leaderlock:
         global cfgleader
         if cfgleader and not isinstance(cfgleader, bool):

confluent_server/confluent/discovery/core.py

@@ -681,6 +681,9 @@ def detected_models():
 async def _recheck_nodes(nodeattribs, configmanager):
     if not cfm.config_is_ready():
         return
+    global rechecklock
+    if rechecklock is None:
+        rechecklock = asyncio.Lock()
     if rechecklock.locked():
         # if already in progress, don't run again
         # it may make sense to schedule a repeat, but will try the easier and less redundant way first
@@ -1643,7 +1646,7 @@ async def newnodes(added, deleting, renamed, configmanager):
 
 rechecker = None
 rechecktime = None
-rechecklock = asyncio.Lock()
+rechecklock = None
 
 async def _periodic_recheck(configmanager):
     global rechecker

confluent_server/confluent/neighutil.py

@@ -31,7 +31,7 @@ neightable = {}
 neightime = 0
 
 
-neighlock = asyncio.Lock()
+neighlock = None
 
 async def _update_neigh():
     global neightable
@@ -93,6 +93,9 @@ async def get_hwaddr(ipaddr):
         ipaddr = socket.inet_pton(socket.AF_INET6, ipaddr)
     elif '.' in ipaddr:
         ipaddr = socket.inet_pton(socket.AF_INET, ipaddr)
+    global neighlock
+    if neighlock is None:
+        neighlock = asyncio.Lock()
     async with neighlock:
         updated = False
         if os.times()[4] > (neightime + 30):

confluent_server/confluent/networking/macmap.py

@@ -487,7 +487,7 @@ async def find_nodeinfo_by_mac(mac, configmanager):
     return None, {'maccount': 0}
 
 
-mapupdating = asyncio.Lock()
+mapupdating = None
 
 
 async def update_macmap(configmanager, impatient=False):
@@ -498,6 +498,9 @@ async def update_macmap(configmanager, impatient=False):
     recheck the cache as results become possible, rather
     than having to wait for the process to complete to interrogate.
     """
+    global mapupdating
+    if mapupdating is None:
+        mapupdating = asyncio.Lock()
     if mapupdating.locked():
         while mapupdating.locked():
             await asyncio.sleep(1)
@@ -530,6 +533,9 @@ async def _full_updatemacmap(configmanager):
     global _macsbyswitch
     global switchbackoff
     start = util.monotonic_time()
+    global mapupdating
+    if mapupdating is None:
+        mapupdating = asyncio.Lock()
     async with mapupdating:
         vintage = util.monotonic_time()
         # Clear all existing entries
@@ -728,6 +734,9 @@ async def handle_read_api_request(pathcomponents, configmanager):
         if len(pathcomponents) == 8:
             return dump_macinfo(pathcomponents[-1])
     elif pathcomponents[2] == 'rescan':
+        global mapupdating
+        if mapupdating is None:
+            mapupdating = asyncio.Lock()
         return [msg.KeyValueData({'scanning': mapupdating.locked()})]
     raise exc.NotFoundException('Unrecognized path {0}'.format(
         '/'.join(pathcomponents)))