diff --git a/confluent_server/confluent/collective/manager.py b/confluent_server/confluent/collective/manager.py index 09aff6db..f628b5a1 100644 --- a/confluent_server/confluent/collective/manager.py +++ b/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 diff --git a/confluent_server/confluent/config/configmanager.py b/confluent_server/confluent/config/configmanager.py index 5d9623cf..4ccedf99 100644 --- a/confluent_server/confluent/config/configmanager.py +++ b/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): diff --git a/confluent_server/confluent/discovery/core.py b/confluent_server/confluent/discovery/core.py index 44599eb8..642e6827 100644 --- a/confluent_server/confluent/discovery/core.py +++ b/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 diff --git a/confluent_server/confluent/neighutil.py b/confluent_server/confluent/neighutil.py index 772eb658..6eb044b5 100644 --- a/confluent_server/confluent/neighutil.py +++ b/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): diff --git a/confluent_server/confluent/networking/macmap.py b/confluent_server/confluent/networking/macmap.py index fb3bc8ff..6ab5e9e3 100644 --- a/confluent_server/confluent/networking/macmap.py +++ b/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)))