我们从Python开源项目中，提取了以下8个代码示例，用于说明如何使用errno.ENOPROTOOPT。
def set_options(self, sock, bound=False): sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) if hasattr(socket, 'SO_REUSEPORT'): # pragma: no cover try: sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) except socket.error as err: if err[0] not in (errno.ENOPROTOOPT, errno.EINVAL): raise if not bound: self.bind(sock) sock.setblocking(0) # make sure that the socket can be inherited if hasattr(sock, "set_inheritable"): sock.set_inheritable(True) sock.listen(self.conf.backlog) return sock
def test_ignore_other_errors(self): inet4_server = self.mox.CreateMock(wsgi_server._SingleAddressWsgiServer) inet6_server = self.mox.CreateMock(wsgi_server._SingleAddressWsgiServer) self.mox.StubOutWithMock(wsgi_server, '_SingleAddressWsgiServer') self.mox.StubOutWithMock(socket, 'getaddrinfo') socket.getaddrinfo('localhost', 0, socket.AF_UNSPEC, socket.SOCK_STREAM, 0, socket.AI_PASSIVE).AndReturn( [(None, None, None, None, ('127.0.0.1', 0, 'baz')), (None, None, None, None, ('::1', 0, 'baz'))]) wsgi_server._SingleAddressWsgiServer(('127.0.0.1', 0), None).AndReturn( inet4_server) inet4_server.start() inet4_server.port = 123 wsgi_server._SingleAddressWsgiServer(('::1', 123), None).AndReturn( inet6_server) inet6_server.start().AndRaise( wsgi_server.BindError('message', (errno.ENOPROTOOPT, 'no protocol'))) self.mox.ReplayAll() self.server.start() self.mox.VerifyAll() self.assertItemsEqual([inet4_server], self.server._servers)
def _init_sock(self): if self.unix_socket: # try remove the sock file it already exists _sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) try: _sock.connect(self.unix_socket) except (socket.error, OSError) as err: if err.args[0] == errno.ECONNREFUSED: os.unlink(self.unix_socket) else: _sock = socket.socket(self.socket_family, socket.SOCK_STREAM) _sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) if hasattr(socket, "SO_REUSEPORT"): try: _sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) except socket.error as err: if err[0] in (errno.ENOPROTOOPT, errno.EINVAL): pass else: raise _sock.settimeout(None) self.sock = _sock
def run(self): self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) if hasattr(socket, "SO_REUSEPORT"): try: self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) except socket.error as le: # RHEL6 defines SO_REUSEPORT but it doesn't work if le.errno == ENOPROTOOPT: pass else: raise addr = socket.inet_aton(SSDP_ADDR) interface = socket.inet_aton('0.0.0.0') cmd = socket.IP_ADD_MEMBERSHIP self.sock.setsockopt(socket.IPPROTO_IP, cmd, addr + interface) self.sock.bind(('0.0.0.0', SSDP_PORT)) self.sock.settimeout(1) while True: try: data, addr = self.sock.recvfrom(1024) self.datagram_received(data, addr) except socket.timeout: continue self.shutdown()
def createInternetSocket(self): skt = Port.createInternetSocket(self) if self.listenMultiple: skt.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) if hasattr(socket, "SO_REUSEPORT"): try: skt.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) except socket.error as le: # RHEL6 defines SO_REUSEPORT but it doesn't work if le.errno == ENOPROTOOPT: pass else: raise return skt
def _makeImageService(self, resource_root): from provisioningserver.rackdservices.image import ( BootImageEndpointService) from twisted.internet.endpoints import AdoptedStreamServerEndpoint port = 5248 # config["port"] # Make a socket with SO_REUSEPORT set so that we can run multiple we # applications. This is easier to do from outside of Twisted as there's # not yet official support for setting socket options. s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) try: s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) except socket_error as e: # Python's socket module was compiled using modern headers # thus defining SO_REUSEPORT would cause issues as it might # running in older kernel that does not support SO_REUSEPORT. # XXX andreserl 2015-04-08 bug=1441684: We need to add a warning # log message when we see this error, and a test for it. if e.errno != ENOPROTOOPT: raise e s.bind(('::', port)) # Use a backlog of 50, which seems to be fairly common. s.listen(50) # Adopt this socket into Twisted's reactor. site_endpoint = AdoptedStreamServerEndpoint( reactor, s.fileno(), s.family) site_endpoint.port = port # Make it easy to get the port number. site_endpoint.socket = s # Prevent garbage collection. image_service = BootImageEndpointService( resource_root=resource_root, endpoint=site_endpoint) image_service.setName("image_service") return image_service
def new_socket(): s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) # SO_REUSEADDR should be equivalent to SO_REUSEPORT for # multicast UDP sockets (p 731, "TCP/IP Illustrated, # Volume 2"), but some BSD-derived systems require # SO_REUSEPORT to be specified explicity. Also, not all # versions of Python have SO_REUSEPORT available. # Catch OSError and socket.error for kernel versions <3.9 because lacking # SO_REUSEPORT support. try: reuseport = socket.SO_REUSEPORT except AttributeError: pass else: try: s.setsockopt(socket.SOL_SOCKET, reuseport, 1) except (OSError, socket.error) as err: # OSError on python 3, socket.error on python 2 if not err.errno == errno.ENOPROTOOPT: raise # OpenBSD needs the ttl and loop values for the IP_MULTICAST_TTL and # IP_MULTICAST_LOOP socket options as an unsigned char. ttl = struct.pack(b'B', 255) s.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, ttl) loop = struct.pack(b'B', 1) s.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, loop) s.bind(('', _MDNS_PORT)) return s
