ports.py (2056B)
1 #!/usr/bin/env python3 2 # -*- coding: utf-8 -*- 3 #*************************************************************************** 4 # _ _ ____ _ 5 # Project ___| | | | _ \| | 6 # / __| | | | |_) | | 7 # | (__| |_| | _ <| |___ 8 # \___|\___/|_| \_\_____| 9 # 10 # Copyright (C) Daniel Stenberg, <daniel@haxx.se>, et al. 11 # 12 # This software is licensed as described in the file COPYING, which 13 # you should have received as part of this distribution. The terms 14 # are also available at https://curl.se/docs/copyright.html. 15 # 16 # You may opt to use, copy, modify, merge, publish, distribute and/or sell 17 # copies of the Software, and permit persons to whom the Software is 18 # furnished to do so, under the terms of the COPYING file. 19 # 20 # This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY 21 # KIND, either express or implied. 22 # 23 # SPDX-License-Identifier: curl 24 # 25 ########################################################################### 26 # 27 import logging 28 import os 29 import socket 30 from collections.abc import Callable 31 from typing import Dict 32 33 from filelock import FileLock 34 35 log = logging.getLogger(__name__) 36 37 38 def alloc_port_set(port_specs: Dict[str, int]) -> Dict[str, int]: 39 socks = [] 40 ports = {} 41 for name, ptype in port_specs.items(): 42 try: 43 s = socket.socket(type=ptype) 44 s.bind(('', 0)) 45 ports[name] = s.getsockname()[1] 46 socks.append(s) 47 except Exception as e: 48 raise e 49 for s in socks: 50 s.close() 51 return ports 52 53 54 def alloc_ports_and_do(port_spec: Dict[str, int], 55 do_func: Callable[[Dict[str, int]], bool], 56 gen_dir, max_tries=1) -> bool: 57 lock_file = os.path.join(gen_dir, 'ports.lock') 58 with FileLock(lock_file): 59 for _ in range(max_tries): 60 port_set = alloc_port_set(port_spec) 61 if do_func(port_set): 62 return True 63 return False