我们从Python开源项目中，提取了以下48个代码示例，用于说明如何使用socket.IPV6_V6ONLY。
def __init__(self, *args, **kwargs): ipv6 = kwargs.pop("ipv6", False) if ipv6: self.address_family = socket.AF_INET6 kwargs["bind_and_activate"] = False else: self.address_family = socket.AF_INET socketserver.TCPServer.__init__(self, *args, **kwargs) if ipv6: # pylint: disable=no-member self.socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1) try: self.server_bind() self.server_activate() except: self.server_close() raise
def init_func(self, creator, address, crypto, crypto_configs, conn_timeout=800, is_ipv6=False): self.__crypto_configs = crypto_configs self.__crypto = crypto self.__conn_timeout = conn_timeout if is_ipv6: fa = socket.AF_INET6 else: fa = socket.AF_INET s = socket.socket(fa, socket.SOCK_STREAM) if is_ipv6: s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.set_socket(s) self.bind(address) self.listen(10) self.register(self.fileno) self.add_evt_read(self.fileno) return self.fileno
def init_func(self, creator, address, crypto, crypto_configs, is_ipv6=False): if is_ipv6: fa = socket.AF_INET6 else: fa = socket.AF_INET s = socket.socket(fa, socket.SOCK_DGRAM) if is_ipv6: s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1) self.set_socket(s) self.bind(address) self.register(self.fileno) self.add_evt_read(self.fileno) self.__encrypt = crypto.encrypt() self.__decrypt = crypto.decrypt() self.__encrypt.config(crypto_configs) self.__decrypt.config(crypto_configs) return self.fileno
def search(timeout): ''' Search for devices implementing WANCommonInterfaceConfig on the network. Search ends the specified number of seconds after the last result (if any) was received. Returns an iterator of root device URLs. ''' with contextlib.ExitStack() as stack: sockets = [] sockets.append(stack.enter_context(socket.socket(socket.AF_INET6, socket.SOCK_DGRAM, socket.IPPROTO_UDP))) sockets.append(stack.enter_context(socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP))) for s in sockets: s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) if s.family == socket.AF_INET6: s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1) with concurrent.futures.ThreadPoolExecutor(len(sockets)) as ex: return itertools.chain.from_iterable(ex.map(lambda s: search_socket(s, timeout, ns['i']), sockets))
def bind(self, family, type, proto=0): """Create (or recreate) the actual socket object.""" self.socket = socket.socket(family, type, proto) prevent_socket_inheritance(self.socket) self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) if self.nodelay and not isinstance(self.bind_addr, str): self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) if self.ssl_adapter is not None: self.socket = self.ssl_adapter.bind(self.socket) # If listening on the IPV6 any address ('::' = IN6ADDR_ANY), # activate dual-stack. See https://bitbucket.org/cherrypy/cherrypy/issue/871. if (hasattr(socket, 'AF_INET6') and family == socket.AF_INET6 and self.bind_addr[0] in ('::', '::0', '::0.0.0.0')): try: self.socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0) except (AttributeError, socket.error): # Apparently, the socket option is not available in # this machine's TCP stack pass self.socket.bind(self.bind_addr)
def bind(self, family, type, proto=0): """Create (or recreate) the actual socket object.""" self.socket = socket.socket(family, type, proto) prevent_socket_inheritance(self.socket) self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) if self.nodelay and not isinstance(self.bind_addr, str): self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) if self.ssl_adapter is not None: self.socket = self.ssl_adapter.bind(self.socket) # If listening on the IPV6 any address ('::' = IN6ADDR_ANY), # activate dual-stack. See http://www.cherrypy.org/ticket/871. if (hasattr(socket, 'AF_INET6') and family == socket.AF_INET6 and self.bind_addr[0] in ('::', '::0', '::0.0.0.0')): try: self.socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0) except (AttributeError, socket.error): # Apparently, the socket option is not available in # this machine's TCP stack pass self.socket.bind(self.bind_addr)
def bind(self, family, type, proto=0): """Create (or recreate) the actual socket object.""" self.socket = socket.socket(family, type, proto) prevent_socket_inheritance(self.socket) self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) if self.nodelay and not isinstance(self.bind_addr, str): self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) if self.ssl_adapter is not None: self.socket = self.ssl_adapter.bind(self.socket) # If listening on the IPV6 any address ('::' = IN6ADDR_ANY), # activate dual-stack. See # https://github.com/cherrypy/cherrypy/issues/871. if (hasattr(socket, 'AF_INET6') and family == socket.AF_INET6 and self.bind_addr[0] in ('::', '::0', '::0.0.0.0')): try: self.socket.setsockopt( socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0) except (AttributeError, socket.error): # Apparently, the socket option is not available in # this machine's TCP stack pass self.socket.bind(self.bind_addr)
def __init__(self, handlers, addrinfo): self.logger = logging.LoggerAdapter(self.logger, dict(context = self)) asyncore.dispatcher.__init__(self) self.handlers = handlers try: af, socktype, proto, canonname, sockaddr = addrinfo # pylint: disable=W0612 self.create_socket(af, socktype) self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) try: self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) except AttributeError: pass if have_ipv6 and af == socket.AF_INET6: self.socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1) self.bind(sockaddr) self.listen(5) except Exception: self.logger.exception("Couldn't set up HTTP listener") self.close() for h in handlers: self.logger.debug("Handling %s", h[0])
def bind_and_activate(self): server_listen_ip = self.orig_server_address[0] try: #???? IPv6 sock = socket.socket(socket.AF_INET6) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) # server_address ??????? v6 ? v4 ???'::' ????????? if server_listen_ip == '': sock.setsockopt(IPPROTO_IPV6, socket.IPV6_V6ONLY, 0) sock.bind(self.orig_server_address) except: sock.close() sock = socket.socket(socket.AF_INET) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind(self.orig_server_address) #???????????? if sys.platform != 'darwin' and server_listen_ip in ('127.0.0.1', '::1'): sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 0) #?? nagle's algorithm ?? sock.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, True) self.socket = sock self.server_address = sock.getsockname() sock.listen(self.request_queue_size) self.is_not_online = False
def _setup_socket(self): self.socket = socket.socket(self.address_family, self.socket_type) if self.allow_reuse_address: self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) if not self.ipv6_only: try: self.socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0) except (AttributeError, socket.error) as e: self.log.debug("Unable to set IPV6_V6ONLY to false %s", e) self.server_bind() self.server_activate()
def init_func(self, creator, address, debug=False, server_side=False, is_ipv6=False): if is_ipv6: fa = socket.AF_INET6 else: fa = socket.AF_INET self.__is_ipv6 = is_ipv6 s = socket.socket(fa, socket.SOCK_DGRAM) if server_side and is_ipv6: s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1) self.set_socket(s) self.__server_side = server_side if server_side: self.bind((address, 53)) else: self.connect((address, 53)) self.__debug = debug self.__host_match = host_match self.__timer = timer.timer() self.__host_match = host_match.host_match() self.set_timeout(self.fileno, self.__LOOP_TIMEOUT) self.register(self.fileno) self.add_evt_read(self.fileno) return self.fileno
def init_func(self, creator, listen, is_ipv6=False): if is_ipv6: fa = socket.AF_INET6 else: fa = socket.AF_INET s = socket.socket(fa, socket.SOCK_STREAM) if is_ipv6: s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.set_socket(s) self.bind(listen) return self.fileno
def bind(family, type, proto): """Create (or recreate) the actual socket object.""" sock = socket.socket(family, type, proto) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True) # If listening on IPv6, activate dual-stack. if family == socket.AF_INET6: sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, False) return sock
def server_bind(self): self.socket.setsockopt(socket.IPPROTO_IP, 15, 1) # IP_FREEBIND if self.__bind_v6only is not None and self.address_family == socket.AF_INET6: self.socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, self.__bind_v6only) super().server_bind()
def set_fwd_socket(self, ip_adr): """ Set classifier forward port base on the ip_adrr version provided :param ip_adr: IP address :type ip: str """ ipver = self._get_current_ip_version(ip_adr) # logger.info('IP version for classifier forward socket is :"%s"', ipver) if ipver == 4: adrr_family = socket.AF_INET elif ipver == 6: adrr_family = socket.AF_INET6 else: adrr_family = socket.AF_INET if self.fwd_socket != None: self.fwd_socket.close() self.fwd_socket = socket.socket(adrr_family, socket.SOCK_DGRAM) self.fwd_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) logger.debug("Forward socket created in classifier with IP %s", ip_adr) # res = self.fwd_socket.getsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY) # logger.info('IPV6_V6ONLY set to :"%s"', res) # self.fwd_socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0) try: self.fwd_socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0) except (AttributeError, socket.error): # Apparently, the socket option is not available in # this machine's TCP stack logger.info("Apparently, the socket option is not available in this machine's TCP stack") pass
def bind(self, family, type, proto=0): """Create (or recreate) the actual socket object.""" self.socket = socket.socket(family, type, proto) prevent_socket_inheritance(self.socket) self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) if self.nodelay: self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) if self.ssl_certificate and self.ssl_private_key: if SSL is None: raise ImportError("You must install pyOpenSSL to use HTTPS.") # See http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/442473 ctx = SSL.Context(SSL.SSLv23_METHOD) ctx.use_privatekey_file(self.ssl_private_key) ctx.use_certificate_file(self.ssl_certificate) self.socket = SSLConnection(ctx, self.socket) self.populate_ssl_environ() # If listening on the IPV6 any address ('::' = IN6ADDR_ANY), # activate dual-stack. See http://www.cherrypy.org/ticket/871. if (not isinstance(self.bind_addr, basestring) and self.bind_addr[0] == '::' and family == socket.AF_INET6): try: self.socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0) except (AttributeError, socket.error): # Apparently, the socket option is not available in # this machine's TCP stack pass self.socket.bind(self.bind_addr)
def _activate(self): address, port = self.url[6:].rsplit(':', 1) port = int(port.rstrip('/')) timeout = max(self.communication_timeout, 30) if address in ('', '0.0.0.0', '::'): try: server_sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) server_sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0) except (AttributeError, socket.error): server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_sock.bind(('', port)) server_sock.listen(1) if not select.select([server_sock], [], [], timeout)[0]: server_sock.close() return False sock, _ = server_sock.accept() server_sock.close() else: if ':' in address: sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) else: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(timeout) sock.connect((address, port)) sock.settimeout(None) self.socket = sock self._first_packet = True return True
def bind(self, family, type, proto=0): """Create (or recreate) the actual socket object.""" self.socket = socket.socket(family, type, proto) prevent_socket_inheritance(self.socket) self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) if self.nodelay and not isinstance(self.bind_addr, str): self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) if self.ssl_adapter is not None: self.socket = self.ssl_adapter.bind(self.socket) host, port = self.bind_addr[:2] # If listening on the IPV6 any address ('::' = IN6ADDR_ANY), # activate dual-stack. See # https://github.com/cherrypy/cherrypy/issues/871. if (hasattr(socket, 'AF_INET6') and family == socket.AF_INET6 and host in ('::', '::0', '::0.0.0.0')): try: self.socket.setsockopt( socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0) except (AttributeError, socket.error): # Apparently, the socket option is not available in # this machine's TCP stack pass self.socket.bind(self.bind_addr)
def server_bind(self): self.socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0) super().server_bind()
def _support_hybrid_ipv6(): """Return True if it is possible to use hybrid IPv6/IPv4 sockets on this platform. """ # Note: IPPROTO_IPV6 constant is broken on Windows, see: # http://bugs.python.org/issue6926 try: if not socket.has_ipv6: return False with contextlib.closing(socket.socket(socket.AF_INET6)) as sock: return not sock.getsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY) except (socket.error, AttributeError): return False
def bind_sockets(port, address=None, family=socket.AF_UNSPEC, backlog=128): """Creates listening sockets bound to the given port and address. Returns a list of socket objects (multiple sockets are returned if the given address maps to multiple IP addresses, which is most common for mixed IPv4 and IPv6 use). Address may be either an IP address or hostname. If it's a hostname, the server will listen on all IP addresses associated with the name. Address may be an empty string or None to listen on all available interfaces. Family may be set to either socket.AF_INET or socket.AF_INET6 to restrict to ipv4 or ipv6 addresses, otherwise both will be used if available. The ``backlog`` argument has the same meaning as for ``socket.listen()``. """ sockets = [] if address == "": address = None flags = socket.AI_PASSIVE if hasattr(socket, "AI_ADDRCONFIG"): # AI_ADDRCONFIG ensures that we only try to bind on ipv6 # if the system is configured for it, but the flag doesn't # exist on some platforms (specifically WinXP, although # newer versions of windows have it) flags |= socket.AI_ADDRCONFIG for res in socket.getaddrinfo(address, port, family, socket.SOCK_STREAM, 0, flags): af, socktype, proto, canonname, sockaddr = res sock = socket.socket(af, socktype, proto) set_close_exec(sock.fileno()) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) if af == socket.AF_INET6: # On linux, ipv6 sockets accept ipv4 too by default, # but this makes it impossible to bind to both # 0.0.0.0 in ipv4 and :: in ipv6. On other systems, # separate sockets *must* be used to listen for both ipv4 # and ipv6. For consistency, always disable ipv4 on our # ipv6 sockets and use a separate ipv4 socket when needed. # # Python 2.x on windows doesn't have IPPROTO_IPV6. if hasattr(socket, "IPPROTO_IPV6"): sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1) sock.setblocking(0) sock.bind(sockaddr) sock.listen(backlog) sockets.append(sock) return sockets
def bind_sockets(port, address=None, family=socket.AF_UNSPEC, backlog=128, flags=None): """Creates listening sockets bound to the given port and address. Returns a list of socket objects (multiple sockets are returned if the given address maps to multiple IP addresses, which is most common for mixed IPv4 and IPv6 use). Address may be either an IP address or hostname. If it's a hostname, the server will listen on all IP addresses associated with the name. Address may be an empty string or None to listen on all available interfaces. Family may be set to either `socket.AF_INET` or `socket.AF_INET6` to restrict to IPv4 or IPv6 addresses, otherwise both will be used if available. The ``backlog`` argument has the same meaning as for `socket.listen() <socket.socket.listen>`. ``flags`` is a bitmask of AI_* flags to `~socket.getaddrinfo`, like ``socket.AI_PASSIVE | socket.AI_NUMERICHOST``. """ sockets = [] if address == "": address = None if not socket.has_ipv6 and family == socket.AF_UNSPEC: # Python can be compiled with --disable-ipv6, which causes # operations on AF_INET6 sockets to fail, but does not # automatically exclude those results from getaddrinfo # results. # http://bugs.python.org/issue16208 family = socket.AF_INET if flags is None: flags = socket.AI_PASSIVE for res in set(socket.getaddrinfo(address, port, family, socket.SOCK_STREAM, 0, flags)): af, socktype, proto, canonname, sockaddr = res try: sock = socket.socket(af, socktype, proto) except socket.error as e: if e.args[0] == errno.EAFNOSUPPORT: continue raise set_close_exec(sock.fileno()) if os.name != 'nt': sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) if af == socket.AF_INET6: # On linux, ipv6 sockets accept ipv4 too by default, # but this makes it impossible to bind to both # 0.0.0.0 in ipv4 and :: in ipv6. On other systems, # separate sockets *must* be used to listen for both ipv4 # and ipv6. For consistency, always disable ipv4 on our # ipv6 sockets and use a separate ipv4 socket when needed. # # Python 2.x on windows doesn't have IPPROTO_IPV6. if hasattr(socket, "IPPROTO_IPV6"): sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1) sock.setblocking(0) sock.bind(sockaddr) sock.listen(backlog) sockets.append(sock) return sockets
def bind_sockets(port, address=None, family=socket.AF_UNSPEC, backlog=128, flags=None): """Creates listening sockets bound to the given port and address. Returns a list of socket objects (multiple sockets are returned if the given address maps to multiple IP addresses, which is most common for mixed IPv4 and IPv6 use). Address may be either an IP address or hostname. If it's a hostname, the server will listen on all IP addresses associated with the name. Address may be an empty string or None to listen on all available interfaces. Family may be set to either `socket.AF_INET` or `socket.AF_INET6` to restrict to IPv4 or IPv6 addresses, otherwise both will be used if available. The ``backlog`` argument has the same meaning as for `socket.listen() <socket.socket.listen>`. ``flags`` is a bitmask of AI_* flags to `~socket.getaddrinfo`, like ``socket.AI_PASSIVE | socket.AI_NUMERICHOST``. """ sockets = [] if address == "": address = None if not socket.has_ipv6 and family == socket.AF_UNSPEC: # Python can be compiled with --disable-ipv6, which causes # operations on AF_INET6 sockets to fail, but does not # automatically exclude those results from getaddrinfo # results. # http://bugs.python.org/issue16208 family = socket.AF_INET if flags is None: flags = socket.AI_PASSIVE for res in set(socket.getaddrinfo(address, port, family, socket.SOCK_STREAM, 0, flags)): af, socktype, proto, canonname, sockaddr = res sock = socket.socket(af, socktype, proto) set_close_exec(sock.fileno()) if os.name != 'nt': sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) if af == socket.AF_INET6: # On linux, ipv6 sockets accept ipv4 too by default, # but this makes it impossible to bind to both # 0.0.0.0 in ipv4 and :: in ipv6. On other systems, # separate sockets *must* be used to listen for both ipv4 # and ipv6. For consistency, always disable ipv4 on our # ipv6 sockets and use a separate ipv4 socket when needed. # # Python 2.x on windows doesn't have IPPROTO_IPV6. if hasattr(socket, "IPPROTO_IPV6"): sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1) sock.setblocking(0) sock.bind(sockaddr) sock.listen(backlog) sockets.append(sock) return sockets
def ftp_EPRT(self, line): """Start an active data channel by choosing the network protocol to use (IPv4/IPv6) as defined in RFC-2428. """ if self._epsvall: self.respond("501 EPRT not allowed after EPSV ALL.") return # Parse EPRT request for getting protocol, IP and PORT. # Request comes in as: # <d>proto<d>ip<d>port<d> # ...where <d> is an arbitrary delimiter character (usually "|") and # <proto> is the network protocol to use (1 for IPv4, 2 for IPv6). try: af, ip, port = line.split(line[0])[1:-1] port = int(port) if not 0 <= port <= 65535: raise ValueError except (ValueError, IndexError, OverflowError): self.respond("501 Invalid EPRT format.") return if af == "1": # test if AF_INET6 and IPV6_V6ONLY if (self.socket.family == socket.AF_INET6 and not SUPPORTS_HYBRID_IPV6): self.respond('522 Network protocol not supported (use 2).') else: try: octs = list(map(int, ip.split('.'))) if len(octs) != 4: raise ValueError for x in octs: if not 0 <= x <= 255: raise ValueError except (ValueError, OverflowError): self.respond("501 Invalid EPRT format.") else: self._make_eport(ip, port) elif af == "2": if self.socket.family == socket.AF_INET: self.respond('522 Network protocol not supported (use 1).') else: self._make_eport(ip, port) else: if self.socket.family == socket.AF_INET: self.respond('501 Unknown network protocol (use 1).') else: self.respond('501 Unknown network protocol (use 2).')
def listener_main(args): """ Totally insecure TCP listener for rpki-rtr protocol. We only implement this because it's all that the routers currently support. In theory, we will all be running TCP-AO in the future, at which point this listener will go away or become a TCP-AO listener. """ # Perhaps we should daemonize? Deal with that later. # server_main() handles args.rpki_rtr_dir. listener = None try: listener = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) listener.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0) except: # pylint: disable=W0702 if listener is not None: listener.close() listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) try: listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) except AttributeError: pass listener.bind(("", args.port)) listener.listen(5) logging.debug("[Listening on port %s]", args.port) while True: try: s, ai = listener.accept() except KeyboardInterrupt: sys.exit(0) logging.debug("[Received connection from %r]", ai) pid = os.fork() if pid == 0: os.dup2(s.fileno(), 0) # pylint: disable=E1103 os.dup2(s.fileno(), 1) # pylint: disable=E1103 s.close() #os.closerange(3, os.sysconf("SC_OPEN_MAX")) server_main(args) sys.exit() else: logging.debug("[Spawned server %d]", pid) while True: try: pid, status = os.waitpid(0, os.WNOHANG) # pylint: disable=W0612 if pid: logging.debug("[Server %s exited]", pid) continue except: # pylint: disable=W0702 pass break
