// spell-checker:disable

pub(crate) use _socket::module_def;

#[cfg(feature = "ssl")]
pub(super) use _socket::{PySocket, SockWaitKind, sock_wait, timeout_error_msg};

#[pymodule]
mod _socket {
    use crate::common::lock::{PyMappedRwLockReadGuard, PyRwLock, PyRwLockReadGuard};
    #[cfg(all(unix, not(target_os = "redox")))]
    use crate::vm::convert::ToPyException;
    use crate::vm::{
        AsObject, Py, PyObjectRef, PyPayload, PyRef, PyResult, VirtualMachine,
        builtins::{
            PyBaseExceptionRef, PyListRef, PyModule, PyOSError, PyStrRef, PyTupleRef, PyTypeRef,
            PyUtf8StrRef,
        },
        convert::{IntoPyException, ToPyObject, TryFromBorrowedObject, TryFromObject},
        function::{
            ArgBytesLike, ArgIntoFloat, ArgMemoryBuffer, ArgStrOrBytesLike, Either, FsPath,
            OptionalArg, OptionalOption,
        },
        types::{Constructor, DefaultConstructor, Destructor, Initializer, Representable},
        utils::ToCString,
    };
    use rustpython_host_env::os::ErrorExt;
    #[cfg(any(unix, windows))]
    use rustpython_host_env::socket as host_socket;
    #[cfg(windows)]
    use rustpython_host_env::windows as host_windows;

    #[expect(clippy::unnecessary_wraps, reason = "Needs to comply with a signature")]
    pub(crate) fn module_exec(vm: &VirtualMachine, module: &Py<PyModule>) -> PyResult<()> {
        #[cfg(windows)]
        host_windows::init_winsock();

        __module_exec(vm, module);
        Ok(())
    }

    use core::{
        mem::MaybeUninit,
        net::{Ipv4Addr, Ipv6Addr, SocketAddr},
        time::Duration,
    };
    use crossbeam_utils::atomic::AtomicCell;
    use host_socket::raw::Socket;
    use num_traits::ToPrimitive;
    use std::{
        ffi,
        io::{self, Read, Write},
        net::{self, Shutdown, ToSocketAddrs},
        time::Instant,
    };

    #[cfg(unix)]
    use libc as c;
    #[cfg(windows)]
    mod c {
        pub(super) use rustpython_host_env::socket::{
            AF_APPLETALK, AF_DECnet, AF_INET, AF_INET6, AF_IPX, AF_LINK, AF_UNSPEC, AI_ADDRCONFIG,
            AI_ALL, AI_CANONNAME, AI_NUMERICHOST, AI_NUMERICSERV, AI_PASSIVE, AI_V4MAPPED,
            EAI_AGAIN, EAI_BADFLAGS, EAI_FAIL, EAI_FAMILY, EAI_MEMORY, EAI_NODATA, EAI_NONAME,
            EAI_SERVICE, EAI_SOCKTYPE, INADDR_ANY, INADDR_BROADCAST, INADDR_LOOPBACK, INADDR_NONE,
            IP_ADD_MEMBERSHIP, IP_DROP_MEMBERSHIP, IP_HDRINCL, IP_MULTICAST_IF, IP_MULTICAST_LOOP,
            IP_MULTICAST_TTL, IP_OPTIONS, IP_RECVDSTADDR, IP_TOS, IP_TTL, IPPORT_RESERVED,
            IPPROTO_AH, IPPROTO_DSTOPTS, IPPROTO_EGP, IPPROTO_ESP, IPPROTO_FRAGMENT, IPPROTO_GGP,
            IPPROTO_HOPOPTS, IPPROTO_ICMP, IPPROTO_ICMPV6, IPPROTO_IDP, IPPROTO_IGMP, IPPROTO_IP,
            IPPROTO_IP as IPPROTO_IPIP, IPPROTO_IPV4, IPPROTO_IPV6, IPPROTO_ND, IPPROTO_NONE,
            IPPROTO_PIM, IPPROTO_PUP, IPPROTO_RAW, IPPROTO_ROUTING, IPPROTO_TCP, IPPROTO_UDP,
            IPV6_CHECKSUM, IPV6_DONTFRAG, IPV6_HOPLIMIT, IPV6_HOPOPTS, IPV6_JOIN_GROUP,
            IPV6_LEAVE_GROUP, IPV6_MULTICAST_HOPS, IPV6_MULTICAST_IF, IPV6_MULTICAST_LOOP,
            IPV6_PKTINFO, IPV6_RECVRTHDR, IPV6_RECVTCLASS, IPV6_RTHDR, IPV6_TCLASS,
            IPV6_UNICAST_HOPS, IPV6_V6ONLY, MSG_BCAST, MSG_CTRUNC, MSG_DONTROUTE, MSG_MCAST,
            MSG_OOB, MSG_PEEK, MSG_TRUNC, MSG_WAITALL, NI_DGRAM, NI_MAXHOST, NI_MAXSERV,
            NI_NAMEREQD, NI_NOFQDN, NI_NUMERICHOST, NI_NUMERICSERV, RCVALL_IPLEVEL, RCVALL_OFF,
            RCVALL_ON, RCVALL_SOCKETLEVELONLY, SD_BOTH as SHUT_RDWR, SD_RECEIVE as SHUT_RD,
            SD_SEND as SHUT_WR, SIO_KEEPALIVE_VALS, SIO_LOOPBACK_FAST_PATH, SIO_RCVALL,
            SO_BROADCAST, SO_ERROR, SO_EXCLUSIVEADDRUSE, SO_KEEPALIVE, SO_LINGER, SO_OOBINLINE,
            SO_RCVBUF, SO_REUSEADDR, SO_SNDBUF, SO_TYPE, SO_USELOOPBACK, SOCK_DGRAM, SOCK_RAW,
            SOCK_RDM, SOCK_SEQPACKET, SOCK_STREAM, SOL_SOCKET, SOMAXCONN, TCP_NODELAY, WSAEBADF,
            WSAENOTSOCK, WSAEWOULDBLOCK, getprotobyname, getservbyname, getservbyport,
        };
    }
    // constants
    #[pyattr(name = "has_ipv6")]
    const HAS_IPV6: bool = true;
    #[pyattr]
    // put IPPROTO_MAX later
    use c::{
        AF_INET, AF_INET6, AF_UNSPEC, INADDR_ANY, INADDR_LOOPBACK, INADDR_NONE, IPPROTO_ICMP,
        IPPROTO_ICMPV6, IPPROTO_IP, IPPROTO_IPV6, IPPROTO_TCP, IPPROTO_TCP as SOL_TCP, IPPROTO_UDP,
        MSG_CTRUNC, MSG_DONTROUTE, MSG_OOB, MSG_PEEK, MSG_TRUNC, MSG_WAITALL, NI_DGRAM, NI_MAXHOST,
        NI_NAMEREQD, NI_NOFQDN, NI_NUMERICHOST, NI_NUMERICSERV, SHUT_RD, SHUT_RDWR, SHUT_WR,
        SO_BROADCAST, SO_ERROR, SO_KEEPALIVE, SO_LINGER, SO_OOBINLINE, SO_RCVBUF, SO_REUSEADDR,
        SO_SNDBUF, SO_TYPE, SOCK_DGRAM, SOCK_STREAM, SOL_SOCKET, TCP_NODELAY,
    };

    #[cfg(not(target_os = "redox"))]
    #[pyattr]
    use c::{
        AF_APPLETALK, AF_DECnet, AF_IPX, IPPROTO_AH, IPPROTO_DSTOPTS, IPPROTO_EGP, IPPROTO_ESP,
        IPPROTO_FRAGMENT, IPPROTO_HOPOPTS, IPPROTO_IDP, IPPROTO_IGMP, IPPROTO_IPIP, IPPROTO_NONE,
        IPPROTO_PIM, IPPROTO_PUP, IPPROTO_RAW, IPPROTO_ROUTING,
    };

    #[cfg(unix)]
    #[pyattr]
    use c::{AF_UNIX, SO_REUSEPORT};

    #[pyattr]
    use c::{AI_ADDRCONFIG, AI_NUMERICHOST, AI_NUMERICSERV, AI_PASSIVE};

    #[cfg(not(target_os = "redox"))]
    #[pyattr]
    use c::{SOCK_RAW, SOCK_RDM, SOCK_SEQPACKET};

    #[cfg(target_os = "android")]
    #[pyattr]
    use c::{SOL_ATALK, SOL_AX25, SOL_IPX, SOL_NETROM, SOL_ROSE};

    #[cfg(target_os = "freebsd")]
    #[pyattr]
    use c::SO_SETFIB;

    #[cfg(target_os = "linux")]
    #[pyattr]
    use c::{
        CAN_BCM, CAN_EFF_FLAG, CAN_EFF_MASK, CAN_ERR_FLAG, CAN_ERR_MASK, CAN_ISOTP, CAN_J1939,
        CAN_RAW, CAN_RAW_ERR_FILTER, CAN_RAW_FD_FRAMES, CAN_RAW_FILTER, CAN_RAW_JOIN_FILTERS,
        CAN_RAW_LOOPBACK, CAN_RAW_RECV_OWN_MSGS, CAN_RTR_FLAG, CAN_SFF_MASK, IPPROTO_MPTCP,
        J1939_IDLE_ADDR, J1939_MAX_UNICAST_ADDR, J1939_NLA_BYTES_ACKED, J1939_NLA_PAD,
        J1939_NO_ADDR, J1939_NO_NAME, J1939_NO_PGN, J1939_PGN_ADDRESS_CLAIMED,
        J1939_PGN_ADDRESS_COMMANDED, J1939_PGN_MAX, J1939_PGN_PDU1_MAX, J1939_PGN_REQUEST,
        SCM_J1939_DEST_ADDR, SCM_J1939_DEST_NAME, SCM_J1939_ERRQUEUE, SCM_J1939_PRIO,
        SO_J1939_ERRQUEUE, SO_J1939_FILTER, SO_J1939_PROMISC, SO_J1939_SEND_PRIO, SOL_CAN_BASE,
        SOL_CAN_RAW,
    };

    // CAN BCM opcodes
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_TX_SETUP: i32 = 1;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_TX_DELETE: i32 = 2;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_TX_READ: i32 = 3;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_TX_SEND: i32 = 4;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_RX_SETUP: i32 = 5;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_RX_DELETE: i32 = 6;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_RX_READ: i32 = 7;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_TX_STATUS: i32 = 8;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_TX_EXPIRED: i32 = 9;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_RX_STATUS: i32 = 10;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_RX_TIMEOUT: i32 = 11;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_RX_CHANGED: i32 = 12;

    // CAN BCM flags (linux/can/bcm.h)
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_SETTIMER: i32 = 0x0001;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_STARTTIMER: i32 = 0x0002;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_TX_COUNTEVT: i32 = 0x0004;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_TX_ANNOUNCE: i32 = 0x0008;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_TX_CP_CAN_ID: i32 = 0x0010;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_RX_FILTER_ID: i32 = 0x0020;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_RX_CHECK_DLC: i32 = 0x0040;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_RX_NO_AUTOTIMER: i32 = 0x0080;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_RX_ANNOUNCE_RESUME: i32 = 0x0100;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_TX_RESET_MULTI_IDX: i32 = 0x0200;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_RX_RTR_FRAME: i32 = 0x0400;
    #[cfg(target_os = "linux")]
    #[pyattr]
    const CAN_BCM_CAN_FD_FRAME: i32 = 0x0800;

    #[cfg(all(target_os = "linux", target_env = "gnu"))]
    #[pyattr]
    use c::SOL_RDS;

    #[cfg(target_os = "netbsd")]
    #[pyattr]
    use c::IPPROTO_VRRP;

    #[cfg(target_vendor = "apple")]
    #[pyattr]
    use c::{AF_SYSTEM, PF_SYSTEM, SYSPROTO_CONTROL, TCP_KEEPALIVE};

    // RFC3542 IPv6 socket options for macOS (netinet6/in6.h)
    // Not available in libc, define manually
    #[cfg(target_vendor = "apple")]
    #[pyattr]
    const IPV6_RECVHOPLIMIT: i32 = 37;
    #[cfg(target_vendor = "apple")]
    #[pyattr]
    const IPV6_RECVRTHDR: i32 = 38;
    #[cfg(target_vendor = "apple")]
    #[pyattr]
    const IPV6_RECVHOPOPTS: i32 = 39;
    #[cfg(target_vendor = "apple")]
    #[pyattr]
    const IPV6_RECVDSTOPTS: i32 = 40;
    #[cfg(target_vendor = "apple")]
    #[pyattr]
    const IPV6_USE_MIN_MTU: i32 = 42;
    #[cfg(target_vendor = "apple")]
    #[pyattr]
    const IPV6_RECVPATHMTU: i32 = 43;
    #[cfg(target_vendor = "apple")]
    #[pyattr]
    const IPV6_PATHMTU: i32 = 44;
    #[cfg(target_vendor = "apple")]
    #[pyattr]
    const IPV6_NEXTHOP: i32 = 48;
    #[cfg(target_vendor = "apple")]
    #[pyattr]
    const IPV6_HOPOPTS: i32 = 49;
    #[cfg(target_vendor = "apple")]
    #[pyattr]
    const IPV6_DSTOPTS: i32 = 50;
    #[cfg(target_vendor = "apple")]
    #[pyattr]
    const IPV6_RTHDR: i32 = 51;
    #[cfg(target_vendor = "apple")]
    #[pyattr]
    const IPV6_RTHDRDSTOPTS: i32 = 57;
    #[cfg(target_vendor = "apple")]
    #[pyattr]
    const IPV6_RTHDR_TYPE_0: i32 = 0;

    #[cfg(windows)]
    #[pyattr]
    use c::{
        IPPORT_RESERVED, IPPROTO_IPV4, RCVALL_IPLEVEL, RCVALL_OFF, RCVALL_ON,
        RCVALL_SOCKETLEVELONLY, SIO_KEEPALIVE_VALS, SIO_LOOPBACK_FAST_PATH, SIO_RCVALL,
        SO_EXCLUSIVEADDRUSE,
    };

    #[cfg(not(windows))]
    #[pyattr]
    const IPPORT_RESERVED: i32 = 1024;

    #[pyattr]
    const IPPORT_USERRESERVED: i32 = 5000;

    #[cfg(any(unix, target_os = "android"))]
    #[pyattr]
    use c::{
        EAI_SYSTEM, MSG_EOR, SO_ACCEPTCONN, SO_DEBUG, SO_DONTROUTE, SO_RCVLOWAT, SO_RCVTIMEO,
        SO_SNDLOWAT, SO_SNDTIMEO,
    };

    #[cfg(any(target_os = "android", target_os = "linux"))]
    #[pyattr]
    use c::{
        ALG_OP_DECRYPT, ALG_OP_ENCRYPT, ALG_SET_AEAD_ASSOCLEN, ALG_SET_AEAD_AUTHSIZE, ALG_SET_IV,
        ALG_SET_KEY, ALG_SET_OP, IP_DEFAULT_MULTICAST_LOOP, IP_RECVOPTS, IP_RETOPTS, IPV6_DSTOPTS,
        IPV6_NEXTHOP, IPV6_PATHMTU, IPV6_RECVDSTOPTS, IPV6_RECVHOPLIMIT, IPV6_RECVHOPOPTS,
        IPV6_RECVPATHMTU, IPV6_RTHDRDSTOPTS, NETLINK_CRYPTO, NETLINK_DNRTMSG, NETLINK_FIREWALL,
        NETLINK_IP6_FW, NETLINK_NFLOG, NETLINK_ROUTE, NETLINK_USERSOCK, NETLINK_XFRM, SO_PASSSEC,
        SO_PEERSEC, SOL_ALG,
    };

    #[cfg(any(target_os = "android", target_vendor = "apple"))]
    #[pyattr]
    use c::{AI_DEFAULT, AI_MASK, AI_V4MAPPED_CFG};

    #[cfg(any(target_os = "freebsd", target_os = "netbsd"))]
    #[pyattr]
    use c::MSG_NOTIFICATION;

    #[cfg(any(target_os = "fuchsia", target_os = "linux"))]
    #[pyattr]
    use c::TCP_USER_TIMEOUT;

    #[cfg(any(unix, target_os = "android", windows))]
    #[pyattr]
    use c::{
        INADDR_BROADCAST, IP_ADD_MEMBERSHIP, IP_DROP_MEMBERSHIP, IP_MULTICAST_IF,
        IP_MULTICAST_LOOP, IP_MULTICAST_TTL, IP_TTL, IPV6_MULTICAST_HOPS, IPV6_MULTICAST_IF,
        IPV6_MULTICAST_LOOP, IPV6_UNICAST_HOPS, IPV6_V6ONLY,
    };

    #[cfg(any(unix, target_os = "android", windows))]
    #[pyattr]
    const INADDR_UNSPEC_GROUP: u32 = 0xe0000000;

    #[cfg(any(unix, target_os = "android", windows))]
    #[pyattr]
    const INADDR_ALLHOSTS_GROUP: u32 = 0xe0000001;

    #[cfg(any(unix, target_os = "android", windows))]
    #[pyattr]
    const INADDR_MAX_LOCAL_GROUP: u32 = 0xe00000ff;

    #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
    #[pyattr]
    use c::{
        AF_ALG, AF_ASH, AF_ATMPVC, AF_ATMSVC, AF_AX25, AF_BRIDGE, AF_CAN, AF_ECONET, AF_IRDA,
        AF_LLC, AF_NETBEUI, AF_NETLINK, AF_NETROM, AF_PACKET, AF_PPPOX, AF_RDS, AF_SECURITY,
        AF_TIPC, AF_VSOCK, AF_WANPIPE, AF_X25, IP_TRANSPARENT, MSG_CONFIRM, MSG_ERRQUEUE,
        MSG_FASTOPEN, MSG_MORE, PF_CAN, PF_PACKET, PF_RDS, SCM_CREDENTIALS, SO_BINDTODEVICE,
        SO_MARK, SOL_IP, SOL_TIPC, SOL_UDP, TCP_CORK, TCP_DEFER_ACCEPT, TCP_LINGER2, TCP_QUICKACK,
        TCP_SYNCNT, TCP_WINDOW_CLAMP,
    };

    // gated on presence of AF_VSOCK:
    #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
    #[pyattr]
    const SO_VM_SOCKETS_BUFFER_SIZE: u32 = 0;

    // gated on presence of AF_VSOCK:
    #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
    #[pyattr]
    const SO_VM_SOCKETS_BUFFER_MIN_SIZE: u32 = 1;

    // gated on presence of AF_VSOCK:
    #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
    #[pyattr]
    const SO_VM_SOCKETS_BUFFER_MAX_SIZE: u32 = 2;

    // gated on presence of AF_VSOCK:
    #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
    #[pyattr]
    const VMADDR_CID_ANY: u32 = 0xffffffff; // 0xffffffff

    // gated on presence of AF_VSOCK:
    #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
    #[pyattr]
    const VMADDR_PORT_ANY: u32 = 0xffffffff; // 0xffffffff

    // gated on presence of AF_VSOCK:
    #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
    #[pyattr]
    const VMADDR_CID_HOST: u32 = 2;

    // gated on presence of AF_VSOCK:
    #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
    #[pyattr]
    const VM_SOCKETS_INVALID_VERSION: u32 = 0xffffffff; // 0xffffffff

    // TODO: gated on https://github.com/rust-lang/libc/pull/1662
    // // gated on presence of AF_VSOCK:
    // #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
    // #[pyattr(name = "IOCTL_VM_SOCKETS_GET_LOCAL_CID", once)]
    // fn ioctl_vm_sockets_get_local_cid(_vm: &VirtualMachine) -> i32 {
    //     c::_IO(7, 0xb9)
    // }

    #[cfg(not(any(target_os = "android", target_os = "fuchsia", target_os = "linux")))]
    #[pyattr]
    const SOL_IP: i32 = 0;

    #[cfg(not(any(target_os = "android", target_os = "fuchsia", target_os = "linux")))]
    #[pyattr]
    const SOL_UDP: i32 = 17;

    #[cfg(any(target_os = "android", target_os = "linux", windows))]
    #[pyattr]
    use c::{IP_OPTIONS, IPV6_HOPOPTS, IPV6_RECVRTHDR, IPV6_RTHDR};

    #[cfg(any(
        target_os = "dragonfly",
        target_os = "freebsd",
        target_vendor = "apple"
    ))]
    #[pyattr]
    use c::{IPPROTO_HELLO, IPPROTO_XTP, LOCAL_PEERCRED, MSG_EOF};

    #[cfg(any(target_os = "netbsd", target_os = "openbsd", windows))]
    #[pyattr]
    use c::{MSG_BCAST, MSG_MCAST};

    #[cfg(any(
        target_os = "android",
        target_os = "fuchsia",
        target_os = "freebsd",
        target_os = "linux"
    ))]
    #[pyattr]
    use c::{IPPROTO_UDPLITE, TCP_CONGESTION};

    #[cfg(any(
        target_os = "android",
        target_os = "fuchsia",
        target_os = "freebsd",
        target_os = "linux"
    ))]
    #[pyattr]
    const UDPLITE_SEND_CSCOV: i32 = 10;

    #[cfg(any(
        target_os = "android",
        target_os = "fuchsia",
        target_os = "freebsd",
        target_os = "linux"
    ))]
    #[pyattr]
    const UDPLITE_RECV_CSCOV: i32 = 11;

    #[cfg(any(
        target_os = "android",
        target_os = "fuchsia",
        target_os = "linux",
        target_os = "openbsd"
    ))]
    #[pyattr]
    use c::AF_KEY;

    #[cfg(any(
        target_os = "android",
        target_os = "fuchsia",
        target_os = "linux",
        target_os = "redox"
    ))]
    #[pyattr]
    use c::SO_DOMAIN;

    #[cfg(any(
        target_os = "android",
        target_os = "fuchsia",
        all(
            target_os = "linux",
            any(
                target_arch = "aarch64",
                target_arch = "x86",
                target_arch = "loongarch64",
                target_arch = "mips",
                target_arch = "powerpc",
                target_arch = "powerpc64",
                target_arch = "riscv64",
                target_arch = "s390x",
                target_arch = "x86_64"
            )
        ),
        target_os = "redox"
    ))]
    #[pyattr]
    use c::SO_PRIORITY;

    #[cfg(any(
        target_os = "dragonfly",
        target_os = "freebsd",
        target_os = "netbsd",
        target_os = "openbsd"
    ))]
    #[pyattr]
    use c::IPPROTO_MOBILE;

    #[cfg(any(
        target_os = "dragonfly",
        target_os = "freebsd",
        target_os = "netbsd",
        target_vendor = "apple"
    ))]
    #[pyattr]
    use c::SCM_CREDS;

    #[cfg(any(
        target_os = "freebsd",
        target_os = "fuchsia",
        target_os = "linux",
        target_vendor = "apple"
    ))]
    #[pyattr]
    use c::TCP_FASTOPEN;

    #[cfg(any(
        target_os = "android",
        target_os = "freebsd",
        target_os = "fuchsia",
        all(
            target_os = "linux",
            any(
                target_arch = "aarch64",
                target_arch = "x86",
                target_arch = "loongarch64",
                target_arch = "mips",
                target_arch = "powerpc",
                target_arch = "powerpc64",
                target_arch = "riscv64",
                target_arch = "s390x",
                target_arch = "x86_64"
            )
        ),
        target_os = "redox"
    ))]
    #[pyattr]
    use c::SO_PROTOCOL;

    #[cfg(any(
        target_os = "android",
        target_os = "dragonfly",
        target_os = "freebsd",
        target_os = "linux",
        target_vendor = "apple",
        windows
    ))]
    #[pyattr]
    use c::IPV6_DONTFRAG;

    #[cfg(any(
        target_os = "android",
        target_os = "dragonfly",
        target_os = "fuchsia",
        target_os = "linux",
        target_os = "redox"
    ))]
    #[pyattr]
    use c::{SO_PASSCRED, SO_PEERCRED};

    #[cfg(any(
        target_os = "android",
        target_os = "freebsd",
        target_os = "fuchsia",
        target_os = "linux",
        target_os = "netbsd"
    ))]
    #[pyattr]
    use c::TCP_INFO;

    #[cfg(any(
        target_os = "android",
        target_os = "freebsd",
        target_os = "fuchsia",
        target_os = "linux",
        target_vendor = "apple"
    ))]
    #[pyattr]
    use c::IP_RECVTOS;

    #[cfg(any(
        target_os = "android",
        target_os = "netbsd",
        target_os = "redox",
        target_vendor = "apple",
        windows
    ))]
    #[pyattr]
    use c::NI_MAXSERV;

    #[cfg(any(
        target_os = "dragonfly",
        target_os = "freebsd",
        target_os = "netbsd",
        target_os = "openbsd",
        target_vendor = "apple"
    ))]
    #[pyattr]
    use c::{IPPROTO_EON, IPPROTO_IPCOMP};

    #[cfg(any(
        target_os = "dragonfly",
        target_os = "freebsd",
        target_os = "netbsd",
        target_vendor = "apple",
        windows
    ))]
    #[pyattr]
    use c::IPPROTO_ND;

    #[cfg(any(
        target_os = "android",
        target_os = "dragonfly",
        target_os = "freebsd",
        target_os = "linux",
        target_vendor = "apple",
        windows
    ))]
    #[pyattr]
    use c::{IPV6_CHECKSUM, IPV6_HOPLIMIT};

    #[cfg(any(
        target_os = "android",
        target_os = "freebsd",
        target_os = "fuchsia",
        target_os = "linux",
        target_os = "netbsd"
    ))]
    #[pyattr]
    use c::IPPROTO_SCTP; // also in windows

    #[cfg(any(
        target_os = "android",
        target_os = "freebsd",
        target_os = "fuchsia",
        target_os = "linux",
        target_vendor = "apple",
        windows
    ))]
    #[pyattr]
    use c::{AI_ALL, AI_V4MAPPED};

    #[cfg(any(
        target_os = "android",
        target_os = "linux",
        target_os = "netbsd",
        target_os = "openbsd",
        target_vendor = "apple",
        windows
    ))]
    #[pyattr]
    use c::EAI_NODATA;

    #[cfg(any(
        target_os = "dragonfly",
        target_os = "freebsd",
        target_os = "netbsd",
        target_os = "openbsd",
        target_vendor = "apple",
        windows
    ))]
    #[pyattr]
    use c::{
        AF_LINK, IP_RECVDSTADDR, IPPROTO_GGP, IPV6_JOIN_GROUP, IPV6_LEAVE_GROUP, SO_USELOOPBACK,
    };

    #[cfg(any(
        target_os = "android",
        target_os = "dragonfly",
        target_os = "freebsd",
        target_os = "fuchsia",
        target_os = "linux",
        target_os = "netbsd",
        target_os = "openbsd"
    ))]
    #[pyattr]
    use c::{MSG_CMSG_CLOEXEC, MSG_NOSIGNAL};

    #[cfg(any(
        target_os = "android",
        target_os = "dragonfly",
        target_os = "freebsd",
        target_os = "fuchsia",
        target_os = "linux",
        target_os = "netbsd",
        target_os = "redox"
    ))]
    #[pyattr]
    use c::TCP_KEEPIDLE;

    #[cfg(any(
        target_os = "android",
        target_os = "dragonfly",
        target_os = "freebsd",
        target_os = "fuchsia",
        target_os = "linux",
        target_os = "netbsd",
        target_vendor = "apple"
    ))]
    #[pyattr]
    use c::{TCP_KEEPCNT, TCP_KEEPINTVL};

    #[cfg(any(
        target_os = "android",
        target_os = "dragonfly",
        target_os = "freebsd",
        target_os = "fuchsia",
        target_os = "linux",
        target_os = "netbsd",
        target_os = "openbsd",
        target_os = "redox"
    ))]
    #[pyattr]
    use c::{SOCK_CLOEXEC, SOCK_NONBLOCK};

    #[cfg(any(
        target_os = "android",
        target_os = "dragonfly",
        target_os = "freebsd",
        target_os = "fuchsia",
        target_os = "linux",
        target_os = "netbsd",
        target_os = "openbsd",
        target_vendor = "apple"
    ))]
    #[pyattr]
    use c::{
        AF_ROUTE, AF_SNA, EAI_OVERFLOW, IPPROTO_GRE, IPPROTO_RSVP, IPPROTO_TP, IPV6_RECVPKTINFO,
        MSG_DONTWAIT, SCM_RIGHTS, TCP_MAXSEG,
    };

    #[cfg(any(
        target_os = "android",
        target_os = "dragonfly",
        target_os = "freebsd",
        target_os = "linux",
        target_os = "netbsd",
        target_os = "openbsd",
        target_vendor = "apple",
        windows
    ))]
    #[pyattr]
    use c::IPV6_PKTINFO;

    #[cfg(any(
        target_os = "android",
        target_os = "freebsd",
        target_os = "fuchsia",
        target_os = "linux",
        target_os = "netbsd",
        target_os = "openbsd",
        target_vendor = "apple",
        windows
    ))]
    #[pyattr]
    use c::AI_CANONNAME;

    #[cfg(any(
        target_os = "android",
        target_os = "dragonfly",
        target_os = "freebsd",
        target_os = "fuchsia",
        target_os = "linux",
        target_os = "netbsd",
        target_os = "openbsd",
        target_vendor = "apple",
        windows
    ))]
    #[pyattr]
    use c::{
        EAI_AGAIN, EAI_BADFLAGS, EAI_FAIL, EAI_FAMILY, EAI_MEMORY, EAI_NONAME, EAI_SERVICE,
        EAI_SOCKTYPE, IP_HDRINCL, IP_TOS, IPV6_RECVTCLASS, IPV6_TCLASS, SOMAXCONN,
    };

    #[cfg(not(any(
        target_os = "android",
        target_os = "dragonfly",
        target_os = "freebsd",
        target_os = "fuchsia",
        target_os = "linux",
        target_os = "netbsd",
        target_os = "openbsd",
        target_vendor = "apple",
        windows
    )))]
    #[pyattr]
    const SOMAXCONN: i32 = 5; // Common value

    // HERE IS WHERE THE BLUETOOTH CONSTANTS START
    // TODO: there should be a more intelligent way of detecting bluetooth on a platform.
    //       CPython uses header-detection, but blocks NetBSD and DragonFly BSD
    #[cfg(any(
        target_os = "android",
        target_os = "freebsd",
        target_os = "fuchsia",
        target_os = "linux",
        target_os = "openbsd"
    ))]
    #[pyattr]
    use c::AF_BLUETOOTH;

    #[cfg(any(
        target_os = "android",
        target_os = "freebsd",
        target_os = "fuchsia",
        target_os = "linux",
        target_os = "openbsd"
    ))]
    #[pyattr]
    const BDADDR_ANY: &str = "00:00:00:00:00:00";
    #[cfg(any(
        target_os = "android",
        target_os = "freebsd",
        target_os = "fuchsia",
        target_os = "linux",
        target_os = "openbsd"
    ))]
    #[pyattr]
    const BDADDR_LOCAL: &str = "00:00:00:FF:FF:FF";
    // HERE IS WHERE THE BLUETOOTH CONSTANTS END

    #[cfg(windows)]
    #[pyattr]
    use host_socket::{
        IPPROTO_CBT, IPPROTO_ICLFXBM, IPPROTO_IGP, IPPROTO_L2TP, IPPROTO_PGM, IPPROTO_RDP,
        IPPROTO_SCTP, IPPROTO_ST,
    };

    #[pyattr]
    fn error(vm: &VirtualMachine) -> PyTypeRef {
        vm.ctx.exceptions.os_error.to_owned()
    }

    #[pyattr]
    fn timeout(vm: &VirtualMachine) -> PyTypeRef {
        vm.ctx.exceptions.timeout_error.to_owned()
    }

    #[pyattr(once)]
    fn herror(vm: &VirtualMachine) -> PyTypeRef {
        vm.ctx.new_exception_type(
            "socket",
            "herror",
            Some(vec![vm.ctx.exceptions.os_error.to_owned()]),
        )
    }
    #[pyattr(once)]
    fn gaierror(vm: &VirtualMachine) -> PyTypeRef {
        vm.ctx.new_exception_type(
            "socket",
            "gaierror",
            Some(vec![vm.ctx.exceptions.os_error.to_owned()]),
        )
    }

    #[pyfunction]
    const fn htonl(x: u32) -> u32 {
        u32::to_be(x)
    }

    #[pyfunction]
    const fn htons(x: u16) -> u16 {
        u16::to_be(x)
    }

    #[pyfunction]
    const fn ntohl(x: u32) -> u32 {
        u32::from_be(x)
    }

    #[pyfunction]
    const fn ntohs(x: u16) -> u16 {
        u16::from_be(x)
    }

    #[cfg(unix)]
    type RawSocket = std::os::unix::io::RawFd;
    #[cfg(windows)]
    type RawSocket = std::os::windows::raw::SOCKET;

    #[cfg(unix)]
    macro_rules! errcode {
        ($e:ident) => {
            c::$e
        };
    }

    #[cfg(windows)]
    macro_rules! errcode {
    ($e:ident) => {
        paste::paste!(c::[<WSA $e>])
    };
}

    fn get_raw_sock(obj: PyObjectRef, vm: &VirtualMachine) -> PyResult<RawSocket> {
        #[cfg(unix)]
        type CastFrom = core::ffi::c_long;
        #[cfg(windows)]
        type CastFrom = core::ffi::c_longlong;

        // should really just be to_index() but test_socket tests the error messages explicitly
        if obj.fast_isinstance(vm.ctx.types.float_type) {
            return Err(vm.new_type_error("integer argument expected, got float"));
        }
        let int = obj
            .try_index_opt(vm)
            .unwrap_or_else(|| Err(vm.new_type_error("an integer is required")))?;
        int.try_to_primitive::<CastFrom>(vm)
            .map(|sock| sock as RawSocket)
    }

    #[cfg(target_os = "linux")]
    #[derive(FromArgs)]
    struct SendmsgAfalgArgs {
        #[pyarg(any, default)]
        msg: Vec<ArgBytesLike>,
        #[pyarg(named)]
        op: u32,
        #[pyarg(named, default)]
        iv: Option<ArgBytesLike>,
        #[pyarg(named, default)]
        assoclen: OptionalArg<isize>,
        #[pyarg(named, default)]
        flags: i32,
    }

    #[pyattr(name = "socket")]
    #[pyattr(name = "SocketType")]
    #[pyclass(name = "socket")]
    #[derive(Debug, PyPayload)]
    pub struct PySocket {
        kind: AtomicCell<i32>,
        family: AtomicCell<i32>,
        proto: AtomicCell<i32>,
        pub(crate) timeout: AtomicCell<f64>,
        sock: PyRwLock<Option<Socket>>,
    }

    const _: () = assert!(core::mem::size_of::<Option<Socket>>() == core::mem::size_of::<Socket>());

    impl Default for PySocket {
        fn default() -> Self {
            Self {
                kind: AtomicCell::default(),
                family: AtomicCell::default(),
                proto: AtomicCell::default(),
                timeout: AtomicCell::new(-1.0),
                sock: PyRwLock::new(None),
            }
        }
    }

    #[cfg(windows)]
    const CLOSED_ERR: i32 = c::WSAENOTSOCK;
    #[cfg(unix)]
    const CLOSED_ERR: i32 = c::EBADF;

    impl Read for &PySocket {
        fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
            (&mut &*self.sock()?).read(buf)
        }
    }

    impl Write for &PySocket {
        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
            (&mut &*self.sock()?).write(buf)
        }

        fn flush(&mut self) -> std::io::Result<()> {
            (&mut &*self.sock()?).flush()
        }
    }

    impl PySocket {
        pub fn sock_opt(&self) -> Option<PyMappedRwLockReadGuard<'_, Socket>> {
            let sock = PyRwLockReadGuard::try_map(self.sock.read(), |sock| sock.as_ref());
            sock.ok()
        }

        pub fn sock(&self) -> io::Result<PyMappedRwLockReadGuard<'_, Socket>> {
            self.sock_opt()
                .ok_or_else(|| io::Error::from_raw_os_error(CLOSED_ERR))
        }

        fn init_inner(
            &self,
            family: i32,
            socket_kind: i32,
            proto: i32,
            sock: Socket,
        ) -> io::Result<()> {
            self.family.store(family);
            // Mask out SOCK_NONBLOCK and SOCK_CLOEXEC flags from stored type
            // to ensure consistent cross-platform behavior
            #[cfg(any(
                target_os = "android",
                target_os = "dragonfly",
                target_os = "freebsd",
                target_os = "fuchsia",
                target_os = "illumos",
                target_os = "linux",
                target_os = "netbsd",
                target_os = "openbsd",
                target_os = "redox"
            ))]
            let masked_kind = socket_kind & !(c::SOCK_NONBLOCK | c::SOCK_CLOEXEC);
            #[cfg(not(any(
                target_os = "android",
                target_os = "dragonfly",
                target_os = "freebsd",
                target_os = "fuchsia",
                target_os = "illumos",
                target_os = "linux",
                target_os = "netbsd",
                target_os = "openbsd",
                target_os = "redox"
            )))]
            let masked_kind = socket_kind;
            self.kind.store(masked_kind);
            self.proto.store(proto);
            let mut s = self.sock.write();
            let sock = s.insert(sock);
            // If SOCK_NONBLOCK is set, use timeout 0 (non-blocking)
            #[cfg(any(
                target_os = "android",
                target_os = "dragonfly",
                target_os = "freebsd",
                target_os = "fuchsia",
                target_os = "illumos",
                target_os = "linux",
                target_os = "netbsd",
                target_os = "openbsd",
                target_os = "redox"
            ))]
            let timeout = if socket_kind & c::SOCK_NONBLOCK != 0 {
                0.0
            } else {
                DEFAULT_TIMEOUT.load()
            };
            #[cfg(not(any(
                target_os = "android",
                target_os = "dragonfly",
                target_os = "freebsd",
                target_os = "fuchsia",
                target_os = "illumos",
                target_os = "linux",
                target_os = "netbsd",
                target_os = "openbsd",
                target_os = "redox"
            )))]
            let timeout = DEFAULT_TIMEOUT.load();
            self.timeout.store(timeout);
            if timeout >= 0.0 {
                sock.set_nonblocking(true)?;
            }
            Ok(())
        }

        /// returns Err(blocking)
        pub fn get_timeout(&self) -> Result<Duration, bool> {
            let timeout = self.timeout.load();
            if timeout > 0.0 {
                Ok(Duration::from_secs_f64(timeout))
            } else {
                Err(timeout != 0.0)
            }
        }

        fn sock_op<F, R>(
            &self,
            vm: &VirtualMachine,
            wait_kind: SockWaitKind,
            f: F,
        ) -> Result<R, IoOrPyException>
        where
            F: FnMut() -> io::Result<R>,
        {
            let timeout = self.get_timeout().ok();
            self.sock_op_timeout_err(vm, wait_kind, timeout, f)
        }

        fn sock_op_timeout_err<F, R>(
            &self,
            vm: &VirtualMachine,
            wait_kind: SockWaitKind,
            timeout: Option<Duration>,
            mut f: F,
        ) -> Result<R, IoOrPyException>
        where
            F: FnMut() -> io::Result<R>,
        {
            let deadline = timeout.map(Deadline::new);

            loop {
                if deadline.is_some() || matches!(wait_kind, SockWaitKind::Connect) {
                    let sock = self.sock()?;
                    sock_wait_deadline(&sock, wait_kind, deadline.as_ref(), vm)?;
                }

                let err = loop {
                    // Detach thread state around the blocking syscall so
                    // stop-the-world can park this thread (e.g. before fork).
                    match vm.allow_threads(&mut f) {
                        Ok(x) => return Ok(x),
                        Err(e) if e.kind() == io::ErrorKind::Interrupted => vm.check_signals()?,
                        Err(e) => break e,
                    }
                };
                if timeout.is_some() && err.kind() == io::ErrorKind::WouldBlock {
                    continue;
                }
                return Err(err.into());
            }
        }

        fn extract_address(
            &self,
            addr: PyObjectRef,
            caller: &str,
            vm: &VirtualMachine,
        ) -> Result<host_socket::raw::SockAddr, IoOrPyException> {
            let family = self.family.load();
            match family {
                #[cfg(unix)]
                c::AF_UNIX => {
                    use crate::vm::function::ArgStrOrBytesLike;
                    use std::os::unix::ffi::OsStrExt;
                    let buf = ArgStrOrBytesLike::try_from_object(vm, addr)?;
                    let bytes = &*buf.borrow_bytes();
                    let path = match &buf {
                        ArgStrOrBytesLike::Buf(_) => ffi::OsStr::from_bytes(bytes).into(),
                        ArgStrOrBytesLike::Str(s) => vm.fsencode(s)?,
                    };
                    host_socket::raw::SockAddr::unix(path)
                        .map_err(|_| vm.new_os_error("AF_UNIX path too long").into())
                }
                c::AF_INET => {
                    let tuple: PyTupleRef = addr.downcast().map_err(|obj| {
                        vm.new_type_error(format!(
                            "{}(): AF_INET address must be tuple, not {}",
                            caller,
                            obj.class().name()
                        ))
                    })?;
                    if tuple.len() != 2 {
                        return Err(vm
                            .new_type_error("AF_INET address must be a pair (host, post)")
                            .into());
                    }
                    let addr = Address::from_tuple(&tuple, vm)?;
                    let mut addr4 = get_addr(vm, addr.host, c::AF_INET)?;
                    match &mut addr4 {
                        SocketAddr::V4(addr4) => {
                            addr4.set_port(addr.port);
                        }
                        SocketAddr::V6(_) => unreachable!(),
                    }
                    Ok(addr4.into())
                }
                c::AF_INET6 => {
                    let tuple: PyTupleRef = addr.downcast().map_err(|obj| {
                        vm.new_type_error(format!(
                            "{}(): AF_INET6 address must be tuple, not {}",
                            caller,
                            obj.class().name()
                        ))
                    })?;
                    match tuple.len() {
                        2..=4 => {}
                        _ => return Err(vm.new_type_error(
                            "AF_INET6 address must be a tuple (host, port[, flowinfo[, scopeid]])",
                        ).into()),
                    }
                    let (addr, flowinfo, scopeid) = Address::from_tuple_ipv6(&tuple, vm)?;
                    let mut addr6 = get_addr(vm, addr.host, c::AF_INET6)?;
                    match &mut addr6 {
                        SocketAddr::V6(addr6) => {
                            addr6.set_port(addr.port);
                            addr6.set_flowinfo(flowinfo);
                            addr6.set_scope_id(scopeid);
                        }
                        SocketAddr::V4(_) => unreachable!(),
                    }
                    Ok(addr6.into())
                }
                #[cfg(target_os = "linux")]
                c::AF_CAN => {
                    let tuple: PyTupleRef = addr.downcast().map_err(|obj| {
                        vm.new_type_error(format!(
                            "{}(): AF_CAN address must be tuple, not {}",
                            caller,
                            obj.class().name()
                        ))
                    })?;
                    if tuple.is_empty() || tuple.len() > 2 {
                        return Err(vm
                            .new_type_error(
                                "AF_CAN address must be a tuple (interface,) or (interface, addr)",
                            )
                            .into());
                    }
                    let interface: PyStrRef = tuple[0].clone().downcast().map_err(|obj| {
                        vm.new_type_error(format!(
                            "{}(): AF_CAN interface must be str, not {}",
                            caller,
                            obj.class().name()
                        ))
                    })?;
                    let interface = interface.try_into_utf8(vm).map_err(IoOrPyException::from)?;
                    let ifname = interface.as_str();

                    // Get interface index
                    let ifindex = if ifname.is_empty() {
                        0 // Bind to all CAN interfaces
                    } else {
                        // Check interface name length (IFNAMSIZ is typically 16)
                        if ifname.len() >= 16 {
                            return Err(vm.new_os_error("interface name too long").into());
                        }
                        let cstr = alloc::ffi::CString::new(ifname)
                            .map_err(|_| vm.new_os_error("invalid interface name"))?;
                        host_socket::if_nametoindex_checked(cstr.as_c_str())? as i32
                    };

                    // Create sockaddr_can
                    let mut storage: c::sockaddr_storage = unsafe { core::mem::zeroed() };
                    let can_addr = &mut storage as *mut c::sockaddr_storage as *mut c::sockaddr_can;
                    unsafe {
                        (*can_addr).can_family = c::AF_CAN as c::sa_family_t;
                        (*can_addr).can_ifindex = ifindex;
                    }
                    let storage: host_socket::raw::SockAddrStorage =
                        unsafe { core::mem::transmute(storage) };
                    Ok(unsafe {
                        host_socket::raw::SockAddr::new(
                            storage,
                            core::mem::size_of::<c::sockaddr_can>() as c::socklen_t,
                        )
                    })
                }
                #[cfg(target_os = "linux")]
                c::AF_ALG => {
                    let tuple: PyTupleRef = addr.downcast().map_err(|obj| {
                        vm.new_type_error(format!(
                            "{}(): AF_ALG address must be tuple, not {}",
                            caller,
                            obj.class().name()
                        ))
                    })?;
                    if tuple.len() != 2 {
                        return Err(vm
                            .new_type_error("AF_ALG address must be a tuple (type, name)")
                            .into());
                    }
                    let alg_type: PyStrRef = tuple[0].clone().downcast().map_err(|obj| {
                        vm.new_type_error(format!(
                            "{}(): AF_ALG type must be str, not {}",
                            caller,
                            obj.class().name()
                        ))
                    })?;
                    let alg_name: PyStrRef = tuple[1].clone().downcast().map_err(|obj| {
                        vm.new_type_error(format!(
                            "{}(): AF_ALG name must be str, not {}",
                            caller,
                            obj.class().name()
                        ))
                    })?;

                    let alg_type = alg_type.try_into_utf8(vm).map_err(IoOrPyException::from)?;
                    let alg_name = alg_name.try_into_utf8(vm).map_err(IoOrPyException::from)?;
                    let type_str = alg_type.as_str();
                    let name_str = alg_name.as_str();

                    // salg_type is 14 bytes, salg_name is 64 bytes
                    if type_str.len() >= 14 {
                        return Err(vm.new_value_error("type too long").into());
                    }
                    if name_str.len() >= 64 {
                        return Err(vm.new_value_error("name too long").into());
                    }

                    // Create sockaddr_alg
                    let mut storage: c::sockaddr_storage = unsafe { core::mem::zeroed() };
                    let alg_addr = &mut storage as *mut c::sockaddr_storage as *mut c::sockaddr_alg;
                    unsafe {
                        (*alg_addr).salg_family = c::AF_ALG as c::sa_family_t;
                        // Copy type string
                        for (i, b) in type_str.bytes().enumerate() {
                            (*alg_addr).salg_type[i] = b;
                        }
                        // Copy name string
                        for (i, b) in name_str.bytes().enumerate() {
                            (*alg_addr).salg_name[i] = b;
                        }
                    }
                    let storage: host_socket::raw::SockAddrStorage =
                        unsafe { core::mem::transmute(storage) };
                    Ok(unsafe {
                        host_socket::raw::SockAddr::new(
                            storage,
                            core::mem::size_of::<c::sockaddr_alg>() as c::socklen_t,
                        )
                    })
                }
                _ => Err(vm.new_os_error(format!("{caller}(): bad family")).into()),
            }
        }

        fn connect_inner(
            &self,
            address: PyObjectRef,
            caller: &str,
            vm: &VirtualMachine,
        ) -> Result<(), IoOrPyException> {
            let sock_addr = self.extract_address(address, caller, vm)?;

            let sock = self.sock()?;
            let err = match vm.allow_threads(|| sock.connect(&sock_addr)) {
                Ok(()) => return Ok(()),
                Err(e) => e,
            };

            let wait_connect = if err.kind() == io::ErrorKind::Interrupted {
                vm.check_signals()?;
                self.timeout.load() != 0.0
            } else {
                #[cfg(unix)]
                use c::EINPROGRESS;
                #[cfg(windows)]
                use c::WSAEWOULDBLOCK as EINPROGRESS;

                self.timeout.load() > 0.0 && err.raw_os_error() == Some(EINPROGRESS)
            };

            if wait_connect {
                self.sock_op(vm, SockWaitKind::Connect, || {
                    let sock = self.sock()?;
                    let err = sock.take_error()?;
                    match err {
                        Some(e) => Err(e),
                        None => Ok(()),
                    }
                })
            } else {
                Err(err.into())
            }
        }
    }

    impl DefaultConstructor for PySocket {}

    #[derive(FromArgs)]
    pub struct SocketInitArgs {
        #[pyarg(any, optional)]
        family: OptionalArg<i32>,
        #[pyarg(any, optional)]
        r#type: OptionalArg<i32>,
        #[pyarg(any, optional)]
        proto: OptionalArg<i32>,
        #[pyarg(any, optional)]
        fileno: OptionalOption<PyObjectRef>,
    }

    impl Initializer for PySocket {
        type Args = SocketInitArgs;

        fn init(zelf: PyRef<Self>, args: Self::Args, vm: &VirtualMachine) -> PyResult<()> {
            Self::_init(zelf, args, vm).map_err(|e| e.into_pyexception(vm))
        }
    }

    impl Representable for PySocket {
        #[inline]
        fn repr_str(zelf: &Py<Self>, _vm: &VirtualMachine) -> PyResult<String> {
            Ok(format!(
                "<socket object, fd={}, family={}, type={}, proto={}>",
                zelf.fileno(),
                zelf.family.load(),
                zelf.kind.load(),
                zelf.proto.load(),
            ))
        }
    }

    impl Destructor for PySocket {
        fn del(zelf: &Py<Self>, vm: &VirtualMachine) -> PyResult<()> {
            // Emit ResourceWarning if socket is still open
            if zelf.sock.read().is_some() {
                let repr = zelf
                    .as_object()
                    .repr(vm)
                    .unwrap_or_else(|_| vm.ctx.new_str("<socket>"));
                let msg = format!("unclosed {}", repr.as_wtf8());
                let _ = crate::vm::warn::warn(
                    vm.ctx.new_str(msg).into(),
                    Some(vm.ctx.exceptions.resource_warning.to_owned()),
                    1,
                    None,
                    vm,
                );
            }
            let _ = zelf.close();
            Ok(())
        }
    }

    #[pyclass(
        with(Constructor, Initializer, Representable, Destructor),
        flags(BASETYPE)
    )]
    impl PySocket {
        fn _init(
            zelf: PyRef<Self>,
            args: <Self as Initializer>::Args,
            vm: &VirtualMachine,
        ) -> Result<(), IoOrPyException> {
            let mut family = args.family.unwrap_or(-1);
            let mut socket_kind = args.r#type.unwrap_or(-1);
            let mut proto = args.proto.unwrap_or(-1);

            if let Ok(audit) = vm.sys_module.get_attr("audit", vm) {
                audit.call(
                    (vm.ctx.new_str("socket.__new__"), family, socket_kind, proto),
                    vm,
                )?;
            }

            let fileno = args.fileno;
            let sock;

            // On Windows, fileno can be bytes from socket.share() for fromshare()
            #[cfg(windows)]
            if let Some(fileno_obj) = fileno.flatten() {
                use crate::vm::builtins::PyBytes;
                if let Ok(bytes) = fileno_obj.clone().downcast::<PyBytes>() {
                    let bytes_data = bytes.as_bytes();
                    let expected_size = host_socket::protocol_info_size();

                    if bytes_data.len() != expected_size {
                        return Err(vm
                            .new_value_error(format!(
                                "socket descriptor string has wrong size, should be {expected_size} bytes"
                            ))
                            .into());
                    }

                    let shared = host_socket::socket_from_share_data(bytes_data)?;
                    family = shared.family;
                    socket_kind = shared.socket_type;
                    proto = shared.protocol;
                    sock = unsafe { sock_from_raw_unchecked(shared.raw as RawSocket) };
                    return Ok(zelf.init_inner(family, socket_kind, proto, sock)?);
                }

                // Not bytes, treat as regular fileno
                let fileno = get_raw_sock(fileno_obj, vm)?;
                sock = sock_from_raw(fileno, vm)?;
                match sock.local_addr() {
                    Ok(addr) if family == -1 => family = addr.family() as i32,
                    Err(e)
                        if family == -1
                            || matches!(
                                e.raw_os_error(),
                                Some(errcode!(ENOTSOCK) | errcode!(EBADF))
                            ) =>
                    {
                        core::mem::forget(sock);
                        return Err(e.into());
                    }
                    _ => {}
                }
                if socket_kind == -1 {
                    socket_kind = sock.r#type().map_err(|e| e.into_pyexception(vm))?.into();
                }
                proto = 0;
                return Ok(zelf.init_inner(family, socket_kind, proto, sock)?);
            }

            #[cfg(not(windows))]
            let fileno = fileno
                .flatten()
                .map(|obj| get_raw_sock(obj, vm))
                .transpose()?;
            #[cfg(not(windows))]
            if let Some(fileno) = fileno {
                sock = sock_from_raw(fileno, vm)?;
                match sock.local_addr() {
                    Ok(addr) if family == -1 => family = addr.family() as i32,
                    Err(e)
                        if family == -1
                            || matches!(e.raw_os_error(), Some(c::ENOTSOCK | c::EBADF)) =>
                    {
                        core::mem::forget(sock);
                        return Err(e.into());
                    }
                    _ => {}
                }
                if socket_kind == -1 {
                    socket_kind = sock.r#type().map_err(|e| e.into_pyexception(vm))?.into();
                }

                cfg_select! {
                    any(
                        target_os = "android",
                        target_os = "freebsd",
                        target_os = "fuchsia",
                        target_os = "linux",
                    ) => {
                        if proto == -1 {
                            proto = sock.protocol()?.map_or(0, Into::into);
                        }
                    }
                    _ => {
                        proto = 0;
                    }
                }

                return Ok(zelf.init_inner(family, socket_kind, proto, sock)?);
            }

            // No fileno provided, create new socket
            {
                if family == -1 {
                    family = c::AF_INET as _
                }
                if socket_kind == -1 {
                    socket_kind = c::SOCK_STREAM
                }
                if proto == -1 {
                    proto = 0
                }
                sock = Socket::new(family.into(), socket_kind.into(), Some(proto.into()))?;
            };
            Ok(zelf.init_inner(family, socket_kind, proto, sock)?)
        }

        #[pymethod]
        fn connect(
            &self,
            address: PyObjectRef,
            vm: &VirtualMachine,
        ) -> Result<(), IoOrPyException> {
            self.connect_inner(address, "connect", vm)
        }

        #[pymethod]
        fn connect_ex(&self, address: PyObjectRef, vm: &VirtualMachine) -> PyResult<i32> {
            match self.connect_inner(address, "connect_ex", vm) {
                Ok(()) => Ok(0),
                Err(err) => err.errno(),
            }
        }

        #[pymethod]
        fn bind(&self, address: PyObjectRef, vm: &VirtualMachine) -> Result<(), IoOrPyException> {
            let sock_addr = self.extract_address(address, "bind", vm)?;

            if let Some(addr) = sock_addr.as_socket()
                && let Ok(audit) = vm.sys_module.get_attr("audit", vm)
            {
                let (ip, port) = match addr {
                    SocketAddr::V4(addr) => (addr.ip().to_string(), addr.port()),
                    SocketAddr::V6(addr) => (addr.ip().to_string(), addr.port()),
                };

                audit.call((vm.ctx.new_str("socket.bind"), (ip, port)), vm)?;
            }

            Ok(self.sock()?.bind(&sock_addr)?)
        }

        #[pymethod]
        fn listen(&self, backlog: OptionalArg<i32>) -> io::Result<()> {
            let backlog = backlog.unwrap_or(128);
            let backlog = if backlog < 0 { 0 } else { backlog };
            self.sock()?.listen(backlog)
        }

        #[pymethod]
        fn _accept(
            &self,
            vm: &VirtualMachine,
        ) -> Result<(RawSocket, PyObjectRef), IoOrPyException> {
            // Use accept_raw() instead of accept() to avoid socket2's set_common_flags()
            // which tries to set SO_NOSIGPIPE and fails with EINVAL on Unix domain sockets on macOS
            let (sock, addr) =
                self.sock_op(vm, SockWaitKind::Read, || self.sock()?.accept_raw())?;
            let fd = into_sock_fileno(sock);
            Ok((fd, get_addr_tuple(&addr, vm)))
        }

        #[pymethod]
        fn recv(
            &self,
            bufsize: usize,
            flags: OptionalArg<i32>,
            vm: &VirtualMachine,
        ) -> Result<Vec<u8>, IoOrPyException> {
            let flags = flags.unwrap_or(0);
            let mut buffer = Vec::with_capacity(bufsize);
            let sock = self.sock()?;
            let n = self.sock_op(vm, SockWaitKind::Read, || {
                sock.recv_with_flags(buffer.spare_capacity_mut(), flags)
            })?;
            unsafe { buffer.set_len(n) };
            Ok(buffer)
        }

        #[pymethod]
        fn recv_into(
            &self,
            buf: ArgMemoryBuffer,
            nbytes: OptionalArg<isize>,
            flags: OptionalArg<i32>,
            vm: &VirtualMachine,
        ) -> Result<usize, IoOrPyException> {
            let flags = flags.unwrap_or(0);
            let sock = self.sock()?;
            let mut buf = buf.borrow_buf_mut();
            let buf = &mut *buf;

            // Handle nbytes parameter
            let read_len = if let OptionalArg::Present(nbytes) = nbytes {
                let nbytes = nbytes
                    .to_usize()
                    .ok_or_else(|| vm.new_value_error("negative buffersize in recv_into"))?;
                nbytes.min(buf.len())
            } else {
                buf.len()
            };

            let buf = &mut buf[..read_len];
            self.sock_op(vm, SockWaitKind::Read, || {
                sock.recv_with_flags(unsafe { slice_as_uninit(buf) }, flags)
            })
        }

        #[pymethod]
        fn recvfrom(
            &self,
            bufsize: isize,
            flags: OptionalArg<i32>,
            vm: &VirtualMachine,
        ) -> Result<(Vec<u8>, PyObjectRef), IoOrPyException> {
            let flags = flags.unwrap_or(0);
            let bufsize = bufsize
                .to_usize()
                .ok_or_else(|| vm.new_value_error("negative buffersize in recvfrom"))?;
            let mut buffer = Vec::with_capacity(bufsize);
            let (n, addr) = self.sock_op(vm, SockWaitKind::Read, || {
                self.sock()?
                    .recv_from_with_flags(buffer.spare_capacity_mut(), flags)
            })?;
            unsafe { buffer.set_len(n) };
            Ok((buffer, get_addr_tuple(&addr, vm)))
        }

        #[pymethod]
        fn recvfrom_into(
            &self,
            buf: ArgMemoryBuffer,
            nbytes: OptionalArg<isize>,
            flags: OptionalArg<i32>,
            vm: &VirtualMachine,
        ) -> Result<(usize, PyObjectRef), IoOrPyException> {
            let mut buf = buf.borrow_buf_mut();
            let buf = &mut *buf;
            let buf = match nbytes {
                OptionalArg::Present(i) => {
                    let i = i.to_usize().ok_or_else(|| {
                        vm.new_value_error("negative buffersize in recvfrom_into")
                    })?;
                    buf.get_mut(..i).ok_or_else(|| {
                        vm.new_value_error("nbytes is greater than the length of the buffer")
                    })?
                }
                OptionalArg::Missing => buf,
            };
            let flags = flags.unwrap_or(0);
            let sock = self.sock()?;
            let (n, addr) = self.sock_op(vm, SockWaitKind::Read, || {
                sock.recv_from_with_flags(unsafe { slice_as_uninit(buf) }, flags)
            })?;
            Ok((n, get_addr_tuple(&addr, vm)))
        }

        #[pymethod]
        fn send(
            &self,
            bytes: ArgBytesLike,
            flags: OptionalArg<i32>,
            vm: &VirtualMachine,
        ) -> Result<usize, IoOrPyException> {
            let flags = flags.unwrap_or(0);
            let buf = bytes.borrow_buf();
            let buf = &*buf;
            self.sock_op(vm, SockWaitKind::Write, || {
                self.sock()?.send_with_flags(buf, flags)
            })
        }

        #[pymethod]
        fn sendall(
            &self,
            bytes: ArgBytesLike,
            flags: OptionalArg<i32>,
            vm: &VirtualMachine,
        ) -> Result<(), IoOrPyException> {
            let flags = flags.unwrap_or(0);

            let timeout = self.get_timeout().ok();

            let deadline = timeout.map(Deadline::new);

            let buf = bytes.borrow_buf();
            let buf = &*buf;
            let mut buf_offset = 0;
            // now we have like 3 layers of interrupt loop :)
            while buf_offset < buf.len() {
                let interval = deadline.as_ref().map(|d| d.time_until()).transpose()?;
                self.sock_op_timeout_err(vm, SockWaitKind::Write, interval, || {
                    let subbuf = &buf[buf_offset..];
                    buf_offset += self.sock()?.send_with_flags(subbuf, flags)?;
                    Ok(())
                })?;
                vm.check_signals()?;
            }
            Ok(())
        }

        #[pymethod]
        fn sendto(
            &self,
            bytes: ArgBytesLike,
            arg2: PyObjectRef,
            arg3: OptionalArg<PyObjectRef>,
            vm: &VirtualMachine,
        ) -> Result<usize, IoOrPyException> {
            // signature is bytes[, flags], address
            let (flags, address) = match arg3 {
                OptionalArg::Present(arg3) => {
                    // should just be i32::try_from_obj but tests check for error message
                    let int = arg2
                        .try_index_opt(vm)
                        .unwrap_or_else(|| Err(vm.new_type_error("an integer is required")))?;
                    let flags = int.try_to_primitive::<i32>(vm)?;
                    (flags, arg3)
                }
                OptionalArg::Missing => (0, arg2),
            };
            let addr = self.extract_address(address, "sendto", vm)?;
            let buf = bytes.borrow_buf();
            let buf = &*buf;
            self.sock_op(vm, SockWaitKind::Write, || {
                self.sock()?.send_to_with_flags(buf, &addr, flags)
            })
        }

        #[cfg(all(unix, not(target_os = "redox")))]
        #[pymethod]
        fn sendmsg(
            &self,
            buffers: Vec<ArgBytesLike>,
            ancdata: OptionalArg,
            flags: OptionalArg<i32>,
            addr: OptionalOption,
            vm: &VirtualMachine,
        ) -> PyResult<usize> {
            let flags = flags.unwrap_or(0);
            let mut msg = host_socket::raw::MsgHdr::new();

            let sockaddr;
            if let Some(addr) = addr.flatten() {
                sockaddr = self
                    .extract_address(addr, "sendmsg", vm)
                    .map_err(|e| e.into_pyexception(vm))?;
                msg = msg.with_addr(&sockaddr);
            }

            let buffers = buffers
                .iter()
                .map(|buf| buf.borrow_buf())
                .collect::<Vec<_>>();
            let buffers = buffers
                .iter()
                .map(|buf| io::IoSlice::new(buf))
                .collect::<Vec<_>>();
            msg = msg.with_buffers(&buffers);

            let control_buf;
            if let OptionalArg::Present(ancdata) = ancdata {
                let cmsgs = vm.extract_elements_with(
                    &ancdata,
                    |obj| -> PyResult<(i32, i32, ArgBytesLike)> {
                        let seq: Vec<PyObjectRef> = obj.try_into_value(vm)?;
                        let [lvl, typ, data]: [PyObjectRef; 3] = seq
                            .try_into()
                            .map_err(|_| vm.new_type_error("expected a sequence of length 3"))?;
                        Ok((
                            lvl.try_into_value(vm)?,
                            typ.try_into_value(vm)?,
                            data.try_into_value(vm)?,
                        ))
                    },
                )?;
                control_buf = Self::pack_cmsgs_to_send(&cmsgs, vm)?;
                if !control_buf.is_empty() {
                    msg = msg.with_control(&control_buf);
                }
            }

            self.sock_op(vm, SockWaitKind::Write, || {
                let sock = self.sock()?;
                sock.sendmsg(&msg, flags)
            })
            .map_err(|e| e.into_pyexception(vm))
        }

        /// sendmsg_afalg([msg], *, op[, iv[, assoclen[, flags]]]) -> int
        ///
        /// Set operation mode and target IV for an AF_ALG socket.
        #[cfg(target_os = "linux")]
        #[pymethod]
        fn sendmsg_afalg(&self, args: SendmsgAfalgArgs, vm: &VirtualMachine) -> PyResult<usize> {
            use std::os::fd::BorrowedFd;

            let msg = args.msg;
            let op = args.op;
            let iv = args.iv;
            let flags = args.flags;

            // Validate assoclen - must be non-negative if provided
            let assoclen: Option<u32> = match args.assoclen {
                OptionalArg::Present(val) if val < 0 => {
                    return Err(vm.new_type_error("assoclen must be non-negative"));
                }
                OptionalArg::Present(val) => Some(val as u32),
                OptionalArg::Missing => None,
            };

            let buffers = msg.iter().map(|buf| buf.borrow_buf()).collect::<Vec<_>>();
            let buffers = buffers
                .iter()
                .map(|buf| io::IoSlice::new(buf))
                .collect::<Vec<_>>();
            let iv = iv.map(|iv| iv.borrow_buf().to_vec());

            self.sock_op(vm, SockWaitKind::Write, || {
                let sock = self.sock()?;
                let fd = unsafe { BorrowedFd::borrow_raw(sock_fileno(&sock)) };
                host_socket::sendmsg_afalg(fd, &buffers, op, iv.as_deref(), assoclen, flags)
            })
            .map_err(|e| e.into_pyexception(vm))
        }

        /// recvmsg(bufsize[, ancbufsize[, flags]]) -> (data, ancdata, msg_flags, address)
        ///
        /// Receive normal data and ancillary data from the socket.
        #[cfg(all(unix, not(target_os = "redox")))]
        #[pymethod]
        fn recvmsg(
            &self,
            bufsize: isize,
            ancbufsize: OptionalArg<isize>,
            flags: OptionalArg<i32>,
            vm: &VirtualMachine,
        ) -> PyResult<PyTupleRef> {
            if bufsize < 0 {
                return Err(vm.new_value_error("negative buffer size in recvmsg"));
            }
            let bufsize = bufsize as usize;

            let ancbufsize = ancbufsize.unwrap_or(0);
            if ancbufsize < 0 {
                return Err(vm.new_value_error("negative ancillary buffer size in recvmsg"));
            }
            let ancbufsize = ancbufsize as usize;
            let flags = flags.unwrap_or(0);

            let msg = self
                .sock_op(vm, SockWaitKind::Read, || {
                    let sock = self.sock()?;
                    let fd = unsafe { std::os::fd::BorrowedFd::borrow_raw(sock_fileno(&sock)) };
                    host_socket::recvmsg(fd, bufsize, ancbufsize, flags)
                })
                .map_err(|e| e.into_pyexception(vm))?;

            // Build ancdata list
            let ancdata = Self::parse_ancillary_data(&msg.ancdata, vm);

            // Build address tuple
            let address = if let Some(address) = msg.address {
                let storage: host_socket::raw::SockAddrStorage =
                    unsafe { core::mem::transmute(address.storage) };
                let addr = unsafe { host_socket::raw::SockAddr::new(storage, address.len as _) };
                get_addr_tuple(&addr, vm)
            } else {
                vm.ctx.none()
            };

            Ok(vm.ctx.new_tuple(vec![
                vm.ctx.new_bytes(msg.data).into(),
                ancdata,
                vm.ctx.new_int(msg.msg_flags).into(),
                address,
            ]))
        }

        /// Parse ancillary data from a received message header
        #[cfg(all(unix, not(target_os = "redox")))]
        fn parse_ancillary_data(
            control: &[host_socket::AncillaryMessage],
            vm: &VirtualMachine,
        ) -> PyObjectRef {
            let mut result = Vec::new();
            for cmsg in control {
                let tuple = vm.ctx.new_tuple(vec![
                    vm.ctx.new_int(cmsg.level).into(),
                    vm.ctx.new_int(cmsg.kind).into(),
                    vm.ctx.new_bytes(cmsg.data.clone()).into(),
                ]);
                result.push(tuple.into());
            }

            vm.ctx.new_list(result).into()
        }

        // based on nix's implementation
        #[cfg(all(unix, not(target_os = "redox")))]
        fn pack_cmsgs_to_send(
            cmsgs: &[(i32, i32, ArgBytesLike)],
            vm: &VirtualMachine,
        ) -> PyResult<Vec<u8>> {
            if cmsgs.is_empty() {
                return Ok(vec![]);
            }
            let data = cmsgs
                .iter()
                .map(|(lvl, typ, buf)| {
                    let data = buf.borrow_buf();
                    (*lvl, *typ, data.to_vec())
                })
                .collect::<Vec<_>>();
            let data_refs = data
                .iter()
                .map(|(lvl, typ, data)| (*lvl, *typ, data.as_slice()))
                .collect::<Vec<_>>();

            host_socket::pack_ancillary_messages(&data_refs).map_err(|err| err.to_pyexception(vm))
        }

        #[pymethod]
        fn close(&self) -> io::Result<()> {
            let sock = self.sock.write().take();
            if let Some(sock) = sock {
                close_inner(into_sock_fileno(sock))?;
            }
            Ok(())
        }

        #[pymethod]
        #[inline]
        fn detach(&self) -> i64 {
            let sock = self.sock.write().take();
            sock.map_or(INVALID_SOCKET as i64, |s| into_sock_fileno(s) as i64)
        }

        #[pymethod]
        fn fileno(&self) -> i64 {
            self.sock
                .read()
                .as_ref()
                .map_or(INVALID_SOCKET as i64, |s| sock_fileno(s) as i64)
        }

        #[pymethod]
        fn getsockname(&self, vm: &VirtualMachine) -> std::io::Result<PyObjectRef> {
            let addr = self.sock()?.local_addr()?;

            Ok(get_addr_tuple(&addr, vm))
        }

        #[pymethod]
        fn getpeername(&self, vm: &VirtualMachine) -> std::io::Result<PyObjectRef> {
            let addr = self.sock()?.peer_addr()?;

            Ok(get_addr_tuple(&addr, vm))
        }

        #[pymethod]
        fn gettimeout(&self) -> Option<f64> {
            let timeout = self.timeout.load();
            if timeout >= 0.0 { Some(timeout) } else { None }
        }

        #[pymethod]
        fn setblocking(&self, block: bool) -> io::Result<()> {
            self.timeout.store(if block { -1.0 } else { 0.0 });
            self.sock()?.set_nonblocking(!block)
        }

        #[pymethod]
        fn getblocking(&self) -> bool {
            self.timeout.load() != 0.0
        }

        #[pymethod]
        fn settimeout(&self, timeout: Option<ArgIntoFloat>, vm: &VirtualMachine) -> PyResult<()> {
            let timeout = match timeout {
                Some(t) => {
                    let f = t.into_float();
                    if f.is_nan() {
                        return Err(vm.new_value_error("Invalid value NaN (not a number)"));
                    }
                    if f < 0.0 || !f.is_finite() {
                        return Err(vm.new_value_error("Timeout value out of range"));
                    }
                    Some(f)
                }
                None => None,
            };
            self.timeout.store(timeout.unwrap_or(-1.0));
            // even if timeout is > 0 the socket needs to be nonblocking in order for us to select() on
            // it
            self.sock()
                .map_err(|e| e.into_pyexception(vm))?
                .set_nonblocking(timeout.is_some())
                .map_err(|e| e.into_pyexception(vm))
        }

        #[pymethod]
        fn getsockopt(
            &self,
            level: i32,
            name: i32,
            buflen: OptionalArg<i32>,
            vm: &VirtualMachine,
        ) -> Result<PyObjectRef, IoOrPyException> {
            let sock = self.sock()?;
            let fd = sock_fileno(&sock);
            let buflen = buflen.unwrap_or(0);
            if buflen == 0 {
                let flag = host_socket::getsockopt_int(fd as _, level, name)?;
                Ok(vm.ctx.new_int(flag).into())
            } else {
                if buflen <= 0 || buflen > 1024 {
                    return Err(vm.new_os_error("getsockopt buflen out of range").into());
                }
                let buf = host_socket::getsockopt_bytes(fd as _, level, name, buflen as usize)?;
                Ok(vm.ctx.new_bytes(buf).into())
            }
        }

        #[pymethod]
        fn setsockopt(
            &self,
            level: i32,
            name: i32,
            value: Option<Either<ArgBytesLike, i32>>,
            optlen: OptionalArg<u32>,
            vm: &VirtualMachine,
        ) -> Result<(), IoOrPyException> {
            let sock = self.sock()?;
            let fd = sock_fileno(&sock);
            match (value, optlen) {
                (Some(Either::A(b)), OptionalArg::Missing) => {
                    b.with_ref(|b| host_socket::setsockopt_bytes(fd as _, level, name, b))?
                }
                (Some(Either::B(val)), OptionalArg::Missing) => {
                    host_socket::setsockopt_int(fd as _, level, name, val)?
                }
                (None, OptionalArg::Present(optlen)) => {
                    host_socket::setsockopt_none(fd as _, level, name, optlen)?
                }
                _ => {
                    return Err(vm
                        .new_type_error("expected the value arg xor the optlen arg")
                        .into());
                }
            }
            Ok(())
        }

        #[pymethod]
        fn shutdown(&self, how: i32, vm: &VirtualMachine) -> Result<(), IoOrPyException> {
            let how = match how {
                c::SHUT_RD => Shutdown::Read,
                c::SHUT_WR => Shutdown::Write,
                c::SHUT_RDWR => Shutdown::Both,
                _ => {
                    return Err(vm
                        .new_value_error("`how` must be SHUT_RD, SHUT_WR, or SHUT_RDWR")
                        .into());
                }
            };
            Ok(self.sock()?.shutdown(how)?)
        }

        #[cfg(windows)]
        #[pymethod]
        fn ioctl(
            &self,
            cmd: PyObjectRef,
            option: PyObjectRef,
            vm: &VirtualMachine,
        ) -> Result<u32, IoOrPyException> {
            use crate::vm::builtins::PyInt;
            use crate::vm::convert::TryFromObject;

            let sock = self.sock()?;
            let fd = sock_fileno(&sock);

            // Convert cmd to u32, returning ValueError for invalid/negative values
            let cmd_int = cmd
                .downcast::<PyInt>()
                .map_err(|_| vm.new_type_error("an integer is required"))?;
            let cmd_val = cmd_int.as_bigint();
            let cmd: u32 = cmd_val
                .to_u32()
                .ok_or_else(|| vm.new_value_error(format!("invalid ioctl command {cmd_val}")))?;

            match cmd {
                c::SIO_RCVALL | c::SIO_LOOPBACK_FAST_PATH => {
                    // Option must be an integer, not None
                    if vm.is_none(&option) {
                        return Err(vm
                            .new_type_error("an integer is required (got type NoneType)")
                            .into());
                    }
                    let option_val: u32 = TryFromObject::try_from_object(vm, option)?;
                    host_socket::ioctl_u32(fd as _, cmd, option_val).map_err(Into::into)
                }
                c::SIO_KEEPALIVE_VALS => {
                    let tuple: PyTupleRef = option
                        .downcast()
                        .map_err(|_| vm.new_type_error("SIO_KEEPALIVE_VALS requires a tuple"))?;
                    if tuple.len() != 3 {
                        return Err(vm
                            .new_type_error(
                                "SIO_KEEPALIVE_VALS requires (onoff, keepalivetime, keepaliveinterval)",
                            )
                            .into());
                    }

                    let ka = host_socket::TcpKeepalive {
                        onoff: TryFromObject::try_from_object(vm, tuple[0].clone())?,
                        keepalivetime: TryFromObject::try_from_object(vm, tuple[1].clone())?,
                        keepaliveinterval: TryFromObject::try_from_object(vm, tuple[2].clone())?,
                    };

                    if cmd != c::SIO_KEEPALIVE_VALS {
                        return Err(vm
                            .new_value_error(format!("invalid ioctl command {cmd}"))
                            .into());
                    }
                    host_socket::ioctl_keepalive(fd as _, ka).map_err(Into::into)
                }
                _ => Err(vm
                    .new_value_error(format!("invalid ioctl command {cmd}"))
                    .into()),
            }
        }

        #[cfg(windows)]
        #[pymethod]
        fn share(&self, process_id: u32, _vm: &VirtualMachine) -> Result<Vec<u8>, IoOrPyException> {
            let sock = self.sock()?;
            let fd = sock_fileno(&sock);
            host_socket::share_socket(fd as _, process_id).map_err(Into::into)
        }

        #[pygetset(name = "type")]
        fn kind(&self) -> i32 {
            self.kind.load()
        }

        #[pygetset]
        fn family(&self) -> i32 {
            self.family.load()
        }

        #[pygetset]
        fn proto(&self) -> i32 {
            self.proto.load()
        }
    }

    struct Address {
        host: PyUtf8StrRef,
        port: u16,
    }

    impl ToSocketAddrs for Address {
        type Iter = alloc::vec::IntoIter<SocketAddr>;
        fn to_socket_addrs(&self) -> io::Result<Self::Iter> {
            (self.host.as_str(), self.port).to_socket_addrs()
        }
    }

    impl TryFromObject for Address {
        fn try_from_object(vm: &VirtualMachine, obj: PyObjectRef) -> PyResult<Self> {
            let tuple = PyTupleRef::try_from_object(vm, obj)?;
            if tuple.len() != 2 {
                Err(vm.new_type_error("Address tuple should have only 2 values"))
            } else {
                Self::from_tuple(&tuple, vm)
            }
        }
    }

    impl Address {
        fn from_tuple(tuple: &[PyObjectRef], vm: &VirtualMachine) -> PyResult<Self> {
            let host = PyStrRef::try_from_object(vm, tuple[0].clone())?;
            let host = host.try_into_utf8(vm)?;
            let port = i32::try_from_borrowed_object(vm, &tuple[1])?;
            let port = port
                .to_u16()
                .ok_or_else(|| vm.new_overflow_error("port must be 0-65535."))?;
            Ok(Self { host, port })
        }

        fn from_tuple_ipv6(
            tuple: &[PyObjectRef],
            vm: &VirtualMachine,
        ) -> PyResult<(Self, u32, u32)> {
            let addr = Self::from_tuple(tuple, vm)?;
            let flowinfo = tuple
                .get(2)
                .map(|obj| obj.clone().try_index(vm)?.try_to_primitive_raw(vm))
                .transpose()?
                .unwrap_or(0);
            let scopeid = tuple
                .get(3)
                .map(|obj| u32::try_from_borrowed_object(vm, obj))
                .transpose()?
                .unwrap_or(0);
            if flowinfo > 0xfffff {
                return Err(vm.new_overflow_error("flowinfo must be 0-1048575."));
            }
            Ok((addr, flowinfo, scopeid))
        }
    }

    fn get_ip_addr_tuple(addr: &SocketAddr, vm: &VirtualMachine) -> PyObjectRef {
        match addr {
            SocketAddr::V4(addr) => (addr.ip().to_string(), addr.port()).to_pyobject(vm),
            SocketAddr::V6(addr) => (
                addr.ip().to_string(),
                addr.port(),
                addr.flowinfo(),
                addr.scope_id(),
            )
                .to_pyobject(vm),
        }
    }

    fn get_addr_tuple(addr: &host_socket::raw::SockAddr, vm: &VirtualMachine) -> PyObjectRef {
        if let Some(addr) = addr.as_socket() {
            return get_ip_addr_tuple(&addr, vm);
        }
        #[cfg(unix)]
        if addr.is_unix() {
            use std::os::unix::ffi::OsStrExt;
            if let Some(abstractpath) = addr.as_abstract_namespace() {
                return vm.ctx.new_bytes([b"\0", abstractpath].concat()).into();
            }
            // necessary on macos
            let path =
                ffi::OsStr::as_bytes(addr.as_pathname().unwrap_or_else(|| "".as_ref()).as_ref());
            let nul_pos = memchr::memchr(b'\0', path).unwrap_or(path.len());
            let path = ffi::OsStr::from_bytes(&path[..nul_pos]);
            return vm.fsdecode(path).into();
        }
        #[cfg(target_os = "linux")]
        {
            let family = addr.family();
            if family == c::AF_CAN as c::sa_family_t {
                // AF_CAN address: (interface_name,) or (interface_name, can_id)
                let can_addr = unsafe { &*(addr.as_ptr() as *const c::sockaddr_can) };
                let ifindex = can_addr.can_ifindex;
                let ifname = if ifindex == 0 {
                    String::new()
                } else {
                    host_socket::if_indextoname_checked(ifindex as u32).unwrap_or_default()
                };
                return vm.ctx.new_tuple(vec![vm.ctx.new_str(ifname).into()]).into();
            }
            if family == c::AF_ALG as c::sa_family_t {
                // AF_ALG address: (type, name)
                let alg_addr = unsafe { &*(addr.as_ptr() as *const c::sockaddr_alg) };
                let type_bytes = &alg_addr.salg_type;
                let name_bytes = &alg_addr.salg_name;
                let type_nul = memchr::memchr(b'\0', type_bytes).unwrap_or(type_bytes.len());
                let name_nul = memchr::memchr(b'\0', name_bytes).unwrap_or(name_bytes.len());
                let type_str = String::from_utf8_lossy(&type_bytes[..type_nul]).into_owned();
                let name_str = String::from_utf8_lossy(&name_bytes[..name_nul]).into_owned();
                return vm
                    .ctx
                    .new_tuple(vec![
                        vm.ctx.new_str(type_str).into(),
                        vm.ctx.new_str(name_str).into(),
                    ])
                    .into();
            }
        }
        // TODO: support more address families
        (String::new(), 0).to_pyobject(vm)
    }

    #[pyfunction]
    fn gethostname(vm: &VirtualMachine) -> PyResult<PyStrRef> {
        if let Ok(audit) = vm.sys_module.get_attr("audit", vm) {
            audit.call((vm.ctx.new_str("socket.gethostname"),), vm)?;
        }

        rustpython_host_env::socket::hostname()
            .into_string()
            .map(|hostname| vm.ctx.new_str(hostname))
            .map_err(|err| vm.new_os_error(err.into_string().unwrap()))
    }

    #[cfg(all(unix, not(any(target_os = "redox", target_os = "android"))))]
    #[pyfunction]
    fn sethostname(hostname: FsPath) -> std::io::Result<()> {
        host_socket::sethostname(hostname.as_bytes())
    }

    #[pyfunction]
    fn inet_aton(ip_string: PyUtf8StrRef, vm: &VirtualMachine) -> PyResult<Vec<u8>> {
        ip_string
            .as_str()
            .parse::<Ipv4Addr>()
            .map(|ip_addr| Vec::<u8>::from(ip_addr.octets()))
            .map_err(|_| vm.new_os_error("illegal IP address string passed to inet_aton"))
    }

    #[pyfunction]
    fn inet_ntoa(packed_ip: ArgBytesLike, vm: &VirtualMachine) -> PyResult<PyStrRef> {
        let packed_ip = packed_ip.borrow_buf();
        let packed_ip = <&[u8; 4]>::try_from(&*packed_ip)
            .map_err(|_| vm.new_os_error("packed IP wrong length for inet_ntoa"))?;
        Ok(vm.ctx.new_str(Ipv4Addr::from(*packed_ip).to_string()))
    }

    fn cstr_opt_as_ptr(x: &OptionalArg<ffi::CString>) -> *const core::ffi::c_char {
        x.as_ref().map_or_else(core::ptr::null, |s| s.as_ptr())
    }

    #[pyfunction]
    fn getservbyname(
        servicename: PyStrRef,
        protocolname: OptionalArg<PyStrRef>,
        vm: &VirtualMachine,
    ) -> PyResult<u16> {
        let cstr_name = servicename.to_cstring(vm)?;
        let cstr_proto = protocolname
            .as_ref()
            .map(|s| s.to_cstring(vm))
            .transpose()?;
        let cstr_proto = cstr_opt_as_ptr(&cstr_proto);
        let serv = unsafe { c::getservbyname(cstr_name.as_ptr() as _, cstr_proto as _) };
        if serv.is_null() {
            return Err(vm.new_os_error("service/proto not found"));
        }
        let port = unsafe { (*serv).s_port };
        Ok(u16::from_be(port as u16))
    }

    #[pyfunction]
    fn getservbyport(
        port: i32,
        protocolname: OptionalArg<PyStrRef>,
        vm: &VirtualMachine,
    ) -> PyResult<String> {
        let port = port
            .to_u16()
            .ok_or_else(|| vm.new_overflow_error("getservbyport: port must be 0-65535."))?;
        let cstr_proto = protocolname
            .as_ref()
            .map(|s| s.to_cstring(vm))
            .transpose()?;
        let cstr_proto = cstr_opt_as_ptr(&cstr_proto);
        let serv = unsafe { c::getservbyport(port.to_be() as _, cstr_proto as _) };
        if serv.is_null() {
            return Err(vm.new_os_error("port/proto not found"));
        }
        let s = unsafe { ffi::CStr::from_ptr((*serv).s_name as _) };
        Ok(s.to_string_lossy().into_owned())
    }

    unsafe fn slice_as_uninit<T>(v: &mut [T]) -> &mut [MaybeUninit<T>] {
        unsafe { &mut *(v as *mut [T] as *mut [MaybeUninit<T>]) }
    }

    enum IoOrPyException {
        Timeout,
        Py(PyBaseExceptionRef),
        Io(io::Error),
    }
    impl From<PyBaseExceptionRef> for IoOrPyException {
        fn from(exc: PyBaseExceptionRef) -> Self {
            Self::Py(exc)
        }
    }
    impl From<PyRef<PyOSError>> for IoOrPyException {
        fn from(exc: PyRef<PyOSError>) -> Self {
            Self::Py(exc.upcast())
        }
    }
    impl From<io::Error> for IoOrPyException {
        fn from(err: io::Error) -> Self {
            Self::Io(err)
        }
    }
    impl IoOrPyException {
        fn errno(self) -> PyResult<i32> {
            match self {
                Self::Timeout => Ok(errcode!(EWOULDBLOCK)),
                Self::Io(err) => Ok(err.posix_errno()),
                Self::Py(exc) => Err(exc),
            }
        }
    }
    impl IntoPyException for IoOrPyException {
        #[inline]
        fn into_pyexception(self, vm: &VirtualMachine) -> PyBaseExceptionRef {
            match self {
                Self::Timeout => timeout_error(vm).upcast(),
                Self::Py(exc) => exc,
                Self::Io(err) => err.into_pyexception(vm),
            }
        }
    }

    #[derive(Copy, Clone)]
    pub(crate) enum SockWaitKind {
        Read,
        Write,
        Connect,
    }

    /// returns Ok(true) on timeout
    #[cfg(feature = "ssl")]
    pub(crate) fn sock_wait(
        sock: &Socket,
        wait_kind: SockWaitKind,
        timeout: Option<Duration>,
        vm: &VirtualMachine,
    ) -> PyResult<bool> {
        match sock_wait_deadline(sock, wait_kind, timeout.map(Deadline::new).as_ref(), vm) {
            Ok(()) => Ok(false),
            Err(IoOrPyException::Timeout) => Ok(true),
            Err(e) => Err(e.into_pyexception(vm)),
        }
    }

    /// returns Err(IoOrPyException::Timeout) on timeout
    fn sock_wait_deadline(
        sock: &Socket,
        wait_kind: SockWaitKind,
        deadline: Option<&Deadline>,
        vm: &VirtualMachine,
    ) -> Result<(), IoOrPyException> {
        #[cfg(unix)]
        {
            use rustpython_host_env::select::{PollFd, poll_fds};

            let mut events = 0;
            if matches!(wait_kind, SockWaitKind::Read) {
                events |= libc::POLLIN | libc::POLLPRI;
            }
            if matches!(wait_kind, SockWaitKind::Write | SockWaitKind::Connect) {
                events |= libc::POLLOUT;
            }
            let mut fds = [PollFd {
                fd: sock_fileno(sock),
                events,
                revents: 0,
            }; 1];

            loop {
                let (timeout, is_capped) = deadline
                    .as_ref()
                    .map(|d| {
                        d.time_until().map(|t| {
                            let timeout_ms = t.as_millis();
                            let is_capped = timeout_ms > i32::MAX as u128;
                            let timeout = if is_capped {
                                i32::MAX
                            } else {
                                timeout_ms as i32
                            };
                            (timeout, is_capped)
                        })
                    })
                    .transpose()?
                    .unwrap_or((-1, false));

                match vm.allow_threads(|| poll_fds(&mut fds, timeout)) {
                    Ok(0) => {
                        if is_capped {
                            continue;
                        }
                        break Err(IoOrPyException::Timeout);
                    }

                    Ok(_) => {
                        if fds[0].revents & libc::POLLNVAL != 0 {
                            break Err(io::Error::from_raw_os_error(libc::EBADF).into());
                        }
                        break Ok(());
                    }

                    Err(e) => {
                        if e.kind() == io::ErrorKind::Interrupted {
                            vm.check_signals()?;
                            continue;
                        }
                        break Err(e.into());
                    }
                }
            }
        }
        #[cfg(windows)]
        {
            use rustpython_host_env::select::{FdSet, select, timeval};

            let fd = sock_fileno(sock) as usize;

            let mut reads = FdSet::new();
            let mut writes = FdSet::new();
            let mut errs = FdSet::new();

            if matches!(wait_kind, SockWaitKind::Read) {
                reads.insert(fd);
                errs.insert(fd);
            }
            if matches!(wait_kind, SockWaitKind::Write | SockWaitKind::Connect) {
                writes.insert(fd);
                errs.insert(fd);
            }

            let mut timeout = deadline
                .as_ref()
                .map(|d| {
                    d.time_until().map(|dur| timeval {
                        tv_sec: dur.as_secs() as _,
                        tv_usec: dur.subsec_micros() as _,
                    })
                })
                .transpose()?;

            match vm.allow_threads(|| {
                select(
                    0, // nfds is ignored on windows
                    &mut reads,
                    &mut writes,
                    &mut errs,
                    timeout.as_mut(),
                )
            }) {
                Ok(0) => Err(IoOrPyException::Timeout),
                Ok(_) => Ok(()),
                Err(e) => Err(e.into()),
            }
        }
    }

    #[derive(FromArgs)]
    struct GAIOptions {
        #[pyarg(positional)]
        host: Option<ArgStrOrBytesLike>,
        #[pyarg(positional)]
        port: Option<Either<ArgStrOrBytesLike, i32>>,

        #[pyarg(positional, default = c::AF_UNSPEC)]
        family: i32,
        #[pyarg(positional, default = 0)]
        ty: i32,
        #[pyarg(positional, default = 0)]
        proto: i32,
        #[pyarg(positional, default = 0)]
        flags: i32,
    }

    #[pyfunction]
    fn getaddrinfo(
        opts: GAIOptions,
        vm: &VirtualMachine,
    ) -> Result<Vec<PyObjectRef>, IoOrPyException> {
        let hints = host_socket::dns::AddrInfoHints {
            socktype: opts.ty,
            protocol: opts.proto,
            address: opts.family,
            flags: opts.flags,
        };

        // Encode host: str uses IDNA encoding, bytes must be valid UTF-8
        let host_encoded: Option<String> = match opts.host.as_ref() {
            Some(ArgStrOrBytesLike::Str(s)) => {
                let encoded =
                    vm.state
                        .codec_registry
                        .encode_text(s.to_owned(), "idna", None, vm)?;
                let host_str = core::str::from_utf8(encoded.as_bytes())
                    .map_err(|_| vm.new_runtime_error("idna output is not utf8"))?;
                Some(host_str.to_owned())
            }
            Some(ArgStrOrBytesLike::Buf(b)) => {
                let bytes = b.borrow_buf();
                let host_str = core::str::from_utf8(&bytes).map_err(|e| {
                    vm.new_unicode_decode_error_real(
                        vm.ctx.new_str("utf-8"),
                        vm.ctx.new_bytes(bytes.to_vec()),
                        e.valid_up_to(),
                        e.error_len().map_or(bytes.len(), |n| e.valid_up_to() + n),
                        vm.ctx.new_str("host bytes is not utf8"),
                    )
                })?;
                Some(host_str.to_owned())
            }
            None => None,
        };
        let host = host_encoded.as_deref();

        // Encode port: str/bytes as service name, int as port number
        let port_encoded: Option<String> = match opts.port.as_ref() {
            Some(Either::A(sb)) => {
                let port_str = match sb {
                    ArgStrOrBytesLike::Str(s) => {
                        // For str, check for surrogates and raise UnicodeEncodeError if found
                        s.to_str()
                            .ok_or_else(|| {
                                let start = s
                                    .as_wtf8()
                                    .code_points()
                                    .position(|c| c.to_char().is_none())
                                    .unwrap();
                                vm.new_unicode_encode_error_real(
                                    vm.ctx.new_str("utf-8"),
                                    (*s).clone(),
                                    start,
                                    start + 1,
                                    vm.ctx.new_str("surrogates not allowed"),
                                )
                            })?
                            .to_owned()
                    }
                    ArgStrOrBytesLike::Buf(b) => {
                        // For bytes, check if it's valid UTF-8
                        let bytes = b.borrow_buf();
                        core::str::from_utf8(&bytes)
                            .map_err(|e| {
                                vm.new_unicode_decode_error_real(
                                    vm.ctx.new_str("utf-8"),
                                    vm.ctx.new_bytes(bytes.to_vec()),
                                    e.valid_up_to(),
                                    e.error_len().map_or(bytes.len(), |n| e.valid_up_to() + n),
                                    vm.ctx.new_str("port is not utf8"),
                                )
                            })?
                            .to_owned()
                    }
                };
                Some(port_str)
            }
            Some(Either::B(i)) => Some(i.to_string()),
            None => None,
        };
        let port = port_encoded.as_deref();

        let addrs = host_socket::dns::getaddrinfo(host, port, Some(hints))
            .map_err(|err| convert_socket_error(vm, err, SocketError::GaiError))?;

        let list = addrs
            .map(|ai| {
                ai.map(|ai| {
                    vm.new_tuple((
                        ai.address,
                        ai.socktype,
                        ai.protocol,
                        ai.canonname,
                        get_ip_addr_tuple(&ai.sockaddr, vm),
                    ))
                    .into()
                })
            })
            .collect::<io::Result<Vec<_>>>()?;
        Ok(list)
    }

    #[pyfunction]
    fn gethostbyaddr(
        addr: PyUtf8StrRef,
        vm: &VirtualMachine,
    ) -> Result<(String, PyListRef, PyListRef), IoOrPyException> {
        let addr = get_addr(vm, addr, c::AF_UNSPEC)?;
        let (hostname, _) = host_socket::dns::getnameinfo(&addr, 0)
            .map_err(|e| convert_socket_error(vm, e, SocketError::HError))?;
        Ok((
            hostname,
            vm.ctx.new_list(vec![]),
            vm.ctx
                .new_list(vec![vm.ctx.new_str(addr.ip().to_string()).into()]),
        ))
    }

    #[pyfunction]
    fn gethostbyname(name: PyUtf8StrRef, vm: &VirtualMachine) -> Result<String, IoOrPyException> {
        let addr = get_addr(vm, name, c::AF_INET)?;
        match addr {
            SocketAddr::V4(ip) => Ok(ip.ip().to_string()),
            _ => unreachable!(),
        }
    }

    #[pyfunction]
    fn gethostbyname_ex(
        name: PyUtf8StrRef,
        vm: &VirtualMachine,
    ) -> Result<(String, PyListRef, PyListRef), IoOrPyException> {
        let addr = get_addr(vm, name, c::AF_INET)?;
        let (hostname, _) = host_socket::dns::getnameinfo(&addr, 0)
            .map_err(|e| convert_socket_error(vm, e, SocketError::HError))?;
        Ok((
            hostname,
            vm.ctx.new_list(vec![]),
            vm.ctx
                .new_list(vec![vm.ctx.new_str(addr.ip().to_string()).into()]),
        ))
    }

    #[pyfunction]
    fn inet_pton(af_inet: i32, ip_string: PyUtf8StrRef, vm: &VirtualMachine) -> PyResult<Vec<u8>> {
        static ERROR_MSG: &str = "illegal IP address string passed to inet_pton";
        let ip_addr = match af_inet {
            c::AF_INET => ip_string
                .as_str()
                .parse::<Ipv4Addr>()
                .map_err(|_| vm.new_os_error(ERROR_MSG.to_owned()))?
                .octets()
                .to_vec(),
            c::AF_INET6 => ip_string
                .as_str()
                .parse::<Ipv6Addr>()
                .map_err(|_| vm.new_os_error(ERROR_MSG.to_owned()))?
                .octets()
                .to_vec(),
            _ => return Err(vm.new_os_error("Address family not supported by protocol")),
        };
        Ok(ip_addr)
    }

    #[pyfunction]
    fn inet_ntop(af_inet: i32, packed_ip: ArgBytesLike, vm: &VirtualMachine) -> PyResult<String> {
        let packed_ip = packed_ip.borrow_buf();
        match af_inet {
            c::AF_INET => {
                let packed_ip = <&[u8; 4]>::try_from(&*packed_ip).map_err(|_| {
                    vm.new_value_error("invalid length of packed IP address string")
                })?;
                Ok(Ipv4Addr::from(*packed_ip).to_string())
            }
            c::AF_INET6 => {
                let packed_ip = <&[u8; 16]>::try_from(&*packed_ip).map_err(|_| {
                    vm.new_value_error("invalid length of packed IP address string")
                })?;
                Ok(get_ipv6_addr_str(Ipv6Addr::from(*packed_ip)))
            }
            _ => Err(vm.new_value_error(format!("unknown address family {af_inet}"))),
        }
    }

    #[pyfunction]
    fn getprotobyname(name: PyStrRef, vm: &VirtualMachine) -> PyResult {
        let cstr = name.to_cstring(vm)?;
        let proto = unsafe { c::getprotobyname(cstr.as_ptr() as _) };
        if proto.is_null() {
            return Err(vm.new_os_error("protocol not found"));
        }
        let num = unsafe { (*proto).p_proto };
        Ok(vm.ctx.new_int(num).into())
    }

    #[pyfunction]
    fn getnameinfo(
        address: PyTupleRef,
        flags: i32,
        vm: &VirtualMachine,
    ) -> Result<(String, String), IoOrPyException> {
        match address.len() {
            2..=4 => {}
            _ => {
                return Err(vm.new_type_error("illegal sockaddr argument").into());
            }
        }
        let (addr, flowinfo, scopeid) = Address::from_tuple_ipv6(&address, vm)?;
        let hints = host_socket::dns::AddrInfoHints {
            address: c::AF_UNSPEC,
            socktype: c::SOCK_DGRAM,
            flags: c::AI_NUMERICHOST,
            protocol: 0,
        };
        let service = addr.port.to_string();
        let host_str = addr.host.as_str();
        let mut res = host_socket::dns::getaddrinfo(Some(host_str), Some(&service), Some(hints))
            .map_err(|e| convert_socket_error(vm, e, SocketError::GaiError))?
            .filter_map(Result::ok);
        let mut ainfo = res.next().unwrap();
        if res.next().is_some() {
            return Err(vm
                .new_os_error("sockaddr resolved to multiple addresses")
                .into());
        }
        match &mut ainfo.sockaddr {
            SocketAddr::V4(_) => {
                if address.len() != 2 {
                    return Err(vm.new_os_error("IPv4 sockaddr must be 2 tuple").into());
                }
            }
            SocketAddr::V6(addr) => {
                addr.set_flowinfo(flowinfo);
                addr.set_scope_id(scopeid);
            }
        }
        host_socket::dns::getnameinfo(&ainfo.sockaddr, flags)
            .map_err(|e| convert_socket_error(vm, e, SocketError::GaiError))
    }

    #[cfg(unix)]
    #[pyfunction]
    fn socketpair(
        family: OptionalArg<i32>,
        socket_kind: OptionalArg<i32>,
        proto: OptionalArg<i32>,
    ) -> Result<(PySocket, PySocket), IoOrPyException> {
        let family = family.unwrap_or(c::AF_UNIX);
        let socket_kind = socket_kind.unwrap_or(c::SOCK_STREAM);
        let proto = proto.unwrap_or(0);
        let (a, b) = Socket::pair(family.into(), socket_kind.into(), Some(proto.into()))?;
        let py_a = PySocket::default();
        py_a.init_inner(family, socket_kind, proto, a)?;
        let py_b = PySocket::default();
        py_b.init_inner(family, socket_kind, proto, b)?;
        Ok((py_a, py_b))
    }

    #[cfg(all(unix, not(target_os = "redox")))]
    type IfIndex = c::c_uint;
    #[cfg(windows)]
    type IfIndex = u32; // NET_IFINDEX but windows-sys 0.59 doesn't have it

    #[cfg(not(target_os = "redox"))]
    #[pyfunction]
    fn if_nametoindex(name: FsPath, vm: &VirtualMachine) -> PyResult<IfIndex> {
        #[cfg(windows)]
        {
            let name = name.to_cstring(vm)?;
            host_socket::if_nametoindex_checked(&name).map_err(|_| vm.new_last_errno_error())
        }
        #[cfg(not(windows))]
        {
            let name = name.to_cstring(vm)?;
            // in case 'if_nametoindex' does not set errno
            rustpython_host_env::os::set_errno(c::ENODEV);
            let ret = unsafe { c::if_nametoindex(name.as_ptr() as _) };
            if ret == 0 {
                Err(vm.new_last_errno_error())
            } else {
                Ok(ret)
            }
        }
    }

    #[cfg(not(target_os = "redox"))]
    #[pyfunction]
    fn if_indextoname(index: IfIndex, vm: &VirtualMachine) -> PyResult<String> {
        #[cfg(windows)]
        {
            host_socket::if_indextoname_checked(index).map_err(|_| vm.new_last_errno_error())
        }
        #[cfg(not(windows))]
        {
            let mut buf = [0; c::IF_NAMESIZE + 1];
            // in case 'if_indextoname' does not set errno
            rustpython_host_env::os::set_errno(c::ENXIO);
            let ret = unsafe { c::if_indextoname(index, buf.as_mut_ptr()) };
            if ret.is_null() {
                Err(vm.new_last_errno_error())
            } else {
                let buf = unsafe { ffi::CStr::from_ptr(buf.as_ptr() as _) };
                Ok(buf.to_string_lossy().into_owned())
            }
        }
    }

    #[cfg(any(
        windows,
        target_os = "dragonfly",
        target_os = "freebsd",
        target_os = "fuchsia",
        target_os = "ios",
        target_os = "linux",
        target_os = "macos",
        target_os = "netbsd",
        target_os = "openbsd",
    ))]
    #[pyfunction]
    fn if_nameindex(vm: &VirtualMachine) -> PyResult<Vec<PyObjectRef>> {
        #[cfg(not(windows))]
        {
            let list = host_socket::if_nameindex()
                .map_err(|err| err.into_pyexception(vm))?
                .into_iter()
                .map(|tup| tup.to_pyobject(vm))
                .collect();

            Ok(list)
        }
        #[cfg(windows)]
        {
            let list = host_socket::if_nameindex()
                .map_err(|err| err.into_pyexception(vm))?
                .into_iter()
                .map(|tup| tup.to_pyobject(vm))
                .collect();
            Ok(list)
        }
    }

    fn get_addr(
        vm: &VirtualMachine,
        pyname: PyUtf8StrRef,
        af: i32,
    ) -> Result<SocketAddr, IoOrPyException> {
        let name = pyname.as_str();
        if name.is_empty() {
            let hints = host_socket::dns::AddrInfoHints {
                address: af,
                socktype: c::SOCK_DGRAM,
                flags: c::AI_PASSIVE,
                protocol: 0,
            };
            let mut res = host_socket::dns::getaddrinfo(None, Some("0"), Some(hints))
                .map_err(|e| convert_socket_error(vm, e, SocketError::GaiError))?;
            let ainfo = res.next().unwrap()?;
            if res.next().is_some() {
                return Err(vm
                    .new_os_error("wildcard resolved to multiple address")
                    .into());
            }
            return Ok(ainfo.sockaddr);
        }
        if name == "255.255.255.255" || name == "<broadcast>" {
            match af {
                c::AF_INET | c::AF_UNSPEC => {}
                _ => {
                    return Err(vm.new_os_error("address family mismatched").into());
                }
            }
            return Ok(SocketAddr::V4(net::SocketAddrV4::new(
                c::INADDR_BROADCAST.into(),
                0,
            )));
        }
        if let c::AF_INET | c::AF_UNSPEC = af
            && let Ok(addr) = name.parse::<Ipv4Addr>()
        {
            return Ok(SocketAddr::V4(net::SocketAddrV4::new(addr, 0)));
        }
        if matches!(af, c::AF_INET | c::AF_UNSPEC)
            && !name.contains('%')
            && let Ok(addr) = name.parse::<Ipv6Addr>()
        {
            return Ok(SocketAddr::V6(net::SocketAddrV6::new(addr, 0, 0, 0)));
        }
        let hints = host_socket::dns::AddrInfoHints {
            address: af,
            ..Default::default()
        };
        let name = vm
            .state
            .codec_registry
            .encode_text(pyname.into_wtf8(), "idna", None, vm)?;
        let name = core::str::from_utf8(name.as_bytes())
            .map_err(|_| vm.new_runtime_error("idna output is not utf8"))?;
        let mut res = host_socket::dns::getaddrinfo(Some(name), None, Some(hints))
            .map_err(|e| convert_socket_error(vm, e, SocketError::GaiError))?;
        Ok(res.next().unwrap().map(|ainfo| ainfo.sockaddr)?)
    }

    fn sock_from_raw(fileno: RawSocket, vm: &VirtualMachine) -> PyResult<Socket> {
        let invalid = cfg_select! {
            windows => fileno == INVALID_SOCKET,
            _ => fileno < 0,
        };
        if invalid {
            return Err(vm.new_value_error("negative file descriptor"));
        }
        Ok(unsafe { sock_from_raw_unchecked(fileno) })
    }
    /// SAFETY: fileno must not be equal to INVALID_SOCKET
    unsafe fn sock_from_raw_unchecked(fileno: RawSocket) -> Socket {
        #[cfg(unix)]
        {
            use std::os::unix::io::FromRawFd;
            unsafe { Socket::from_raw_fd(fileno) }
        }
        #[cfg(windows)]
        {
            use std::os::windows::io::FromRawSocket;
            unsafe { Socket::from_raw_socket(fileno) }
        }
    }
    pub(super) fn sock_fileno(sock: &Socket) -> RawSocket {
        #[cfg(unix)]
        {
            use std::os::unix::io::AsRawFd;
            sock.as_raw_fd()
        }
        #[cfg(windows)]
        {
            use std::os::windows::io::AsRawSocket;
            sock.as_raw_socket()
        }
    }
    fn into_sock_fileno(sock: Socket) -> RawSocket {
        #[cfg(unix)]
        {
            use std::os::unix::io::IntoRawFd;
            sock.into_raw_fd()
        }
        #[cfg(windows)]
        {
            use std::os::windows::io::IntoRawSocket;
            sock.into_raw_socket()
        }
    }

    pub(super) const INVALID_SOCKET: RawSocket = {
        #[cfg(unix)]
        {
            -1
        }
        #[cfg(windows)]
        {
            host_socket::INVALID_RAW_SOCKET as RawSocket
        }
    };

    fn convert_socket_error(
        vm: &VirtualMachine,
        err: host_socket::dns::LookupError,
        err_kind: SocketError,
    ) -> IoOrPyException {
        if let host_socket::dns::LookupErrorKind::System = err.kind() {
            return io::Error::from(err).into();
        }
        let strerr = {
            #[cfg(unix)]
            {
                match err_kind {
                    SocketError::GaiError => host_socket::gai_error_string(err.error_num()),
                    SocketError::HError => host_socket::h_error_string(err.error_num()),
                }
            }
            #[cfg(windows)]
            {
                "getaddrinfo failed".to_owned()
            }
        };
        let exception_cls = match err_kind {
            SocketError::GaiError => gaierror(vm),
            SocketError::HError => herror(vm),
        };
        vm.new_os_subtype_error(exception_cls, Some(err.error_num()), strerr)
            .into()
    }

    fn timeout_error(vm: &VirtualMachine) -> PyRef<PyOSError> {
        timeout_error_msg(vm, "timed out".to_owned())
    }
    pub(crate) fn timeout_error_msg(vm: &VirtualMachine, msg: String) -> PyRef<PyOSError> {
        vm.new_os_subtype_error(timeout(vm), None, msg)
    }

    fn get_ipv6_addr_str(ipv6: Ipv6Addr) -> String {
        match ipv6.to_ipv4() {
            // instead of "::0.0.ddd.ddd" it's "::xxxx"
            Some(v4) if !ipv6.is_unspecified() && matches!(v4.octets(), [0, 0, _, _]) => {
                format!("::{:x}", u32::from(v4))
            }
            _ => ipv6.to_string(),
        }
    }

    pub(crate) struct Deadline {
        deadline: Instant,
    }

    impl Deadline {
        fn new(timeout: Duration) -> Self {
            Self {
                deadline: Instant::now() + timeout,
            }
        }
        fn time_until(&self) -> Result<Duration, IoOrPyException> {
            self.deadline
                .checked_duration_since(Instant::now())
                // past the deadline already
                .ok_or(IoOrPyException::Timeout)
        }
    }

    static DEFAULT_TIMEOUT: AtomicCell<f64> = AtomicCell::new(-1.0);

    #[pyfunction]
    fn getdefaulttimeout() -> Option<f64> {
        let timeout = DEFAULT_TIMEOUT.load();
        if timeout >= 0.0 { Some(timeout) } else { None }
    }

    #[pyfunction]
    fn setdefaulttimeout(timeout: Option<ArgIntoFloat>, vm: &VirtualMachine) -> PyResult<()> {
        let val = match timeout {
            Some(t) => {
                let f = t.into_float();
                if f.is_nan() {
                    return Err(vm.new_value_error("Invalid value NaN (not a number)"));
                }
                if f < 0.0 || !f.is_finite() {
                    return Err(vm.new_value_error("Timeout value out of range"));
                }
                f
            }
            None => -1.0,
        };
        DEFAULT_TIMEOUT.store(val);
        Ok(())
    }

    #[pyfunction]
    fn dup(x: PyObjectRef, vm: &VirtualMachine) -> Result<RawSocket, IoOrPyException> {
        let sock = get_raw_sock(x, vm)?;
        let sock = core::mem::ManuallyDrop::new(sock_from_raw(sock, vm)?);
        let newsock = sock.try_clone()?;
        let fd = into_sock_fileno(newsock);
        #[cfg(windows)]
        host_socket::set_socket_inheritable(fd as _, false)?;
        Ok(fd)
    }

    #[pyfunction]
    fn close(x: PyObjectRef, vm: &VirtualMachine) -> Result<(), IoOrPyException> {
        Ok(close_inner(get_raw_sock(x, vm)?)?)
    }

    fn close_inner(x: RawSocket) -> io::Result<()> {
        host_socket::close_socket_ignore_connreset(x as _)
    }

    enum SocketError {
        HError,
        GaiError,
    }

    #[cfg(all(unix, not(target_os = "redox")))]
    #[pyfunction(name = "CMSG_LEN")]
    fn cmsg_len(length: PyObjectRef, vm: &VirtualMachine) -> PyResult<usize> {
        let length = length.try_index(vm)?.try_to_primitive_raw(vm)?;
        host_socket::checked_cmsg_len(length)
            .ok_or_else(|| vm.new_overflow_error("CMSG_LEN() argument out of range"))
    }

    #[cfg(all(unix, not(target_os = "redox")))]
    #[pyfunction(name = "CMSG_SPACE")]
    fn cmsg_space(length: PyObjectRef, vm: &VirtualMachine) -> PyResult<usize> {
        let length = length.try_index(vm)?.try_to_primitive_raw(vm)?;
        host_socket::checked_cmsg_space(length)
            .ok_or_else(|| vm.new_overflow_error("CMSG_SPACE() argument out of range"))
    }
}
