Python fcntl.fcntl方法代码示例

本文整理汇总了Python中fcntl.fcntl方法的典型用法代码示例。如果您正苦于以下问题:Python fcntl.fcntl方法的具体用法?Python fcntl.fcntl怎么用?Python fcntl.fcntl使用的例子?那么恭喜您, 这里精选的方法代码示例或许可以为您提供帮助。您也可以进一步了解该方法所在模块fcntl的用法示例。

在下文中一共展示了fcntl.fcntl方法的26个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: _set_cloexec

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def _set_cloexec(self):
        """Set the CLOEXEC flag on all open files (except stdin/out/err).
        If self.max_cloexec_files is an integer (the default), then on
        platforms which support it, it represents the max open files setting
        for the operating system. This function will be called just before
        the process is restarted via os.execv() to prevent open files
        from persisting into the new process.
        Set self.max_cloexec_files to 0 to disable this behavior.
        """
        for fd in range(3, self.max_cloexec_files):  # skip stdin/out/err
            try:
                flags = fcntl.fcntl(fd, fcntl.F_GETFD)
            except IOError:
                continue
            fcntl.fcntl(fd, fcntl.F_SETFD, flags | fcntl.FD_CLOEXEC) 
开发者ID:cherrypy,项目名称:cherrypy,代码行数:19,代码来源:wspbus.py


示例2: mac_set_ip_address

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def mac_set_ip_address(self, dev, ip, serverip, netmask):
		ifr = struct.pack('<16sBBHIIIBBHIIIBBHIII',
			self.iface_name,
			16, socket.AF_INET, 0, struct.unpack('<L', socket.inet_pton(socket.AF_INET, ip))[0], 0, 0,
			16, socket.AF_INET, 0, struct.unpack('<L', socket.inet_pton(socket.AF_INET, serverip))[0], 0, 0,
			16, 0, 0, struct.unpack('<L', socket.inet_pton(socket.AF_INET, "255.255.255.255"))[0], 0, 0)
		try:
			sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
			fcntl.ioctl(sock, self.IOCTL_MACOSX_SIOCAIFADDR, ifr)
		except Exception as e:
			common.internal_print("Something went wrong with setting up the interface.", -1)
			print(e)
			sys.exit(-1)
		# adding new route for forwarding packets properly.
		integer_ip = struct.unpack(">I", socket.inet_pton(socket.AF_INET, serverip))[0]
		rangeip = socket.inet_ntop(socket.AF_INET, struct.pack(">I", integer_ip & ((2**int(netmask))-1)<<32-int(netmask)))
		ps = subprocess.Popen(["route", "add", "-net", rangeip+"/"+netmask, serverip], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
		(stdout, stderr) = ps.communicate()
		if stderr:
			if not "File exists" in stderr:
				common.internal_print("Error: adding client route: {0}".format(stderr), -1)
				sys.exit(-1)
		return 
开发者ID:earthquake,项目名称:XFLTReaT,代码行数:27,代码来源:interface.py


示例3: freebsd_tun_alloc

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def freebsd_tun_alloc(self, dev, flags):
		try:
			sockfd = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
			ifr = struct.pack('<16si', 'tun', 0)
			self.iface_name = fcntl.ioctl(sockfd, self.IOCTL_FREEBSD_SIOCIFCREATE2, ifr)
			self.iface_name = self.iface_name.rstrip("\x00")
			buff = array.array('c', dev+"\x00")
			caddr_t, _ = buff.buffer_info()
			ifr = struct.pack('16sP', self.iface_name, caddr_t);
			fcntl.ioctl(sockfd, self.IOCTL_FREEBSD_SIOCSIFNAME, ifr)
			tun = os.open("/dev/"+self.iface_name, os.O_RDWR | os.O_NONBLOCK)
			self.iface_name = dev
		except IOError as e:
			print e
			common.internal_print("Error: Cannot create tunnel. Is {0} in use?".format(dev), -1)
			sys.exit(-1)
		return tun 
开发者ID:earthquake,项目名称:XFLTReaT,代码行数:23,代码来源:interface.py


示例4: read_char_no_blocking

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def read_char_no_blocking():
    ''' Read a character in nonblocking mode, if no characters are present in the buffer, return an empty string '''
    fd = sys.stdin.fileno()
    old_settings = termios.tcgetattr(fd)
    old_flags = fcntl.fcntl(fd, fcntl.F_GETFL)
    try:
        tty.setraw(fd, termios.TCSADRAIN)
        fcntl.fcntl(fd, fcntl.F_SETFL, old_flags | os.O_NONBLOCK)
        return sys.stdin.read(1)
    except IOError as e:
        ErrorNumber = e[0]
        # IOError with ErrorNumber 11(35 in Mac)  is thrown when there is nothing to read(Resource temporarily unavailable)
        if (sys.platform.startswith("linux") and ErrorNumber != 11) or (sys.platform == "darwin" and ErrorNumber != 35):
            raise
        return ""
    finally:
        fcntl.fcntl(fd, fcntl.F_SETFL, old_flags)
        termios.tcsetattr(fd, termios.TCSADRAIN, old_settings) 
开发者ID:pindexis,项目名称:marker,代码行数:20,代码来源:readchar.py


示例5: train

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def train(self):
        i=0
        if(self.args.ipython):
            import fcntl
            fd = sys.stdin.fileno()
            fl = fcntl.fcntl(fd, fcntl.F_GETFL)
            fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
        while((i < self.total_steps or self.total_steps == -1) and not self.gan.destroy):
            i+=1
            start_time = time.time()
            self.step()
            GlobalViewer.tick()
            if (self.args.save_every != None and
                self.args.save_every != -1 and
                self.args.save_every > 0 and
                i % self.args.save_every == 0):
                print(" |= Saving network")
                self.gan.save(self.save_file)
            if self.args.ipython:
                self.check_stdin()
            end_time = time.time() 
开发者ID:HyperGAN,项目名称:HyperGAN,代码行数:25,代码来源:cli.py


示例6: blocking

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def blocking(self, port, mode=False):
        """
        Set port function mode blocking/nonblocking
        :param port: port to set mode
        :param mode: False to set nonblock mode, True for block mode
        """
        fd = self._open([port])[0]
        try:
            fl = fcntl.fcntl(fd, fcntl.F_GETFL)
            if not mode:
                fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
            else:
                fcntl.fcntl(fd, fcntl.F_SETFL, fl & ~os.O_NONBLOCK)
        except Exception as inst:
            print("FAIL: Setting (non)blocking mode: " + str(inst))
            return
        if mode:
            print("PASS: set to blocking mode")
        else:
            print("PASS: set to nonblocking mode") 
开发者ID:avocado-framework,项目名称:avocado-vt,代码行数:26,代码来源:virtio_console_guest.py


示例7: set_fd_flag

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def set_fd_flag(fd, flag):
    """Set a flag on a file descriptor.
    ``fd`` is the file descriptor or file object, ``flag`` the flag as integer.
    """
    flags = fcntl.fcntl(fd, fcntl.F_GETFD, 0)
    fcntl.fcntl(fd, fcntl.F_SETFD, flags | flag) 
开发者ID:mbusb,项目名称:multibootusb,代码行数:10,代码来源:pipe.py


示例8: set_fd_status_flag

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def set_fd_status_flag(fd, flag):
    """Set a status flag on a file descriptor.
    ``fd`` is the file descriptor or file object, ``flag`` the flag as integer.
    """
    flags = fcntl.fcntl(fd, fcntl.F_GETFL, 0)
    fcntl.fcntl(fd, fcntl.F_SETFL, flags | flag) 
开发者ID:mbusb,项目名称:multibootusb,代码行数:10,代码来源:pipe.py


示例9: lin_init

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def lin_init(self):
		global fcntl
		import fcntl 
开发者ID:earthquake,项目名称:XFLTReaT,代码行数:5,代码来源:interface.py


示例10: lin_tun_alloc

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def lin_tun_alloc(self, dev, flags):
		try:
			tun = os.open(Interface.LINUX_CLONEDEV, os.O_RDWR|os.O_NONBLOCK, 0)
			ifr = struct.pack('16sH', dev, flags)
			fcntl.ioctl(tun, self.IOCTL_LINUX_TUNSETIFF, ifr)
		except IOError:
			common.internal_print("Error: Cannot create tunnel. Is {0} in use?".format(dev), -1)
			sys.exit(-1)
		return tun 
开发者ID:earthquake,项目名称:XFLTReaT,代码行数:13,代码来源:interface.py


示例11: lin_set_ip_address

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def lin_set_ip_address(self, dev, ip, serverip, netmask):
		sockfd = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
		try:
			# set IP
			ifr  = struct.pack('<16sH2s4s8s', dev, socket.AF_INET, "\x00"*2, socket.inet_aton(ip), "\x00"*8)
			fcntl.ioctl(sockfd, self.IOCTL_LINUX_SIOCSIFADDR, ifr)
			# get flags
			ifr = struct.pack('<16sh', dev, 0)
			flags = struct.unpack('<16sh', fcntl.ioctl(sockfd, self.IOCTL_LINUX_SIOCSIFFLAGS, ifr))[1]
			# set new flags
			flags = flags | self.IOCTL_LINUX_IFF_UP
			ifr = struct.pack('<16sh', dev, flags)
			# iface up
			fcntl.ioctl(sockfd, self.IOCTL_LINUX_SIOCSIFFLAGS, ifr)
		except Exception as e:
			common.internal_print("Something went wrong with setting up the interface.", -1)
			print(e)
			sys.exit(-1)
		# adding new route for forwarding packets properly.
		integer_ip = struct.unpack(">I", socket.inet_pton(socket.AF_INET, serverip))[0]
		rangeip = socket.inet_ntop(socket.AF_INET, struct.pack(">I", integer_ip & ((2**int(netmask))-1)<<32-int(netmask)))
		integer_netmask = struct.pack(">I", ((2**int(netmask))-1)<<32-int(netmask))
		netmask = socket.inet_ntoa(integer_netmask)
		ps = subprocess.Popen(["route", "add", "-net", rangeip, "netmask", netmask, "dev", dev], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
		(stdout, stderr) = ps.communicate()
		if stderr:
			if not "File exists" in stderr:
				common.internal_print("Error: adding client route: {0}".format(stderr), -1)
				sys.exit(-1)
		return 
开发者ID:earthquake,项目名称:XFLTReaT,代码行数:39,代码来源:interface.py


示例12: lin_set_mtu

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def lin_set_mtu(self, dev, mtu):
		s = socket.socket(type=socket.SOCK_DGRAM)
		try:
			ifr = struct.pack('<16sH', dev, mtu) + '\x00'*14
			fcntl.ioctl(s, self.IOCTL_LINUX_SIOCSIFMTU, ifr)
		except Exception as e:
			common.internal_print("Cannot set MTU ({0}) on interface".format(mtu), -1)
			sys.exit(-1)
		return 
开发者ID:earthquake,项目名称:XFLTReaT,代码行数:12,代码来源:interface.py


示例13: mac_init

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def mac_init(self):
		global fcntl
		import fcntl 
开发者ID:earthquake,项目名称:XFLTReaT,代码行数:5,代码来源:interface.py


示例14: mac_set_mtu

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def mac_set_mtu(self, dev, mtu):
		s = socket.socket(type=socket.SOCK_DGRAM)
		try:
			ifr = struct.pack('<16sH', self.iface_name, 1350)+'\x00'*14
			fcntl.ioctl(s, self.IOCTL_MACOSX_SIOCSIFMTU, ifr)
		except Exception as e:
			common.internal_print("Cannot set MTU ({0}) on interface".format(mtu), -1)
			sys.exit(-1)
		return 
开发者ID:earthquake,项目名称:XFLTReaT,代码行数:12,代码来源:interface.py


示例15: freebsd_init

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def freebsd_init(self):
		global fcntl
		import fcntl
	# @sghctoma for the win. Thanks for the help to support FreeBSD. 
开发者ID:earthquake,项目名称:XFLTReaT,代码行数:7,代码来源:interface.py


示例16: freebsd_set_ip_address

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def freebsd_set_ip_address(self, dev, ip, serverip, netmask):
		sockfd = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
		try:
			#set ip, serverip and netmask
			ifaliasreq = struct.pack('<16sBBHI8sBBHI8sBBHI8sI', self.iface_name,
				16, socket.AF_INET, 0, struct.unpack('<I', socket.inet_aton(ip))[0], '\x00'*8,
				16, socket.AF_INET, 0, struct.unpack('<I', socket.inet_aton(serverip))[0], '\x00'*8,
				16, socket.AF_INET, 0, struct.unpack('<I', socket.inet_aton('255.255.255.255'))[0],
				'\x00'*8,
				0)
			fcntl.ioctl(sockfd, self.IOCTL_FREEBSD_SIOCAIFADDR, ifaliasreq)
			# get flags
			ifr = struct.pack('<16sh', self.iface_name, 0)
			flags = struct.unpack('<16sh', fcntl.ioctl(sockfd, self.IOCTL_FREEBSD_SIOCGIFFLAGS, ifr))[1]
			# set new flags
			flags = flags | self.IOCTL_FREEBSD_IFF_UP
			ifr = struct.pack('<16sh', self.iface_name, flags)
			# iface up
			fcntl.ioctl(sockfd, self.IOCTL_FREEBSD_SIOCSIFFLAGS, ifr)
		except Exception as e:
			common.internal_print("Something went wrong with setting up the interface.", -1)
			print(e)
			sys.exit(-1)
		# adding new route for forwarding packets properly.
		integer_ip = struct.unpack(">I", socket.inet_pton(socket.AF_INET, serverip))[0]
		rangeip = socket.inet_ntop(socket.AF_INET, struct.pack(">I", integer_ip & ((2**int(netmask))-1)<<32-int(netmask)))
		ps = subprocess.Popen(["route", "add", "-net", rangeip+"/"+netmask, serverip], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
		(stdout, stderr) = ps.communicate()
		if stderr:
			if not "File exists" in stderr:
				common.internal_print("Error: adding client route: {0}".format(stderr), -1)
				sys.exit(-1)
		return 
开发者ID:earthquake,项目名称:XFLTReaT,代码行数:41,代码来源:interface.py


示例17: freebsd_close_tunnel

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def freebsd_close_tunnel(self, tun):
		try:
			os.close(tun)
		except:
			pass
		s = socket.socket(type=socket.SOCK_DGRAM)
		try:
			ifr = struct.pack('<16s', self.iface_name) + '\x00'*16
			fcntl.ioctl(s, self.IOCTL_FREEBSD_SIOCIFDESTROY, ifr)
		except Exception as e:
			common.internal_print("Cannot destroy interface: {0}".format(dev), -1)
		return 
开发者ID:earthquake,项目名称:XFLTReaT,代码行数:16,代码来源:interface.py


示例18: SetFileNonBlocking

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def SetFileNonBlocking(f, non_blocking=True):
  """Set non-blocking flag on a file object.
  Args:
    f: file
    non_blocking: bool, default True, non-blocking mode or not
  """
  flags = fcntl.fcntl(f.fileno(), fcntl.F_GETFL)
  if bool(flags & os.O_NONBLOCK) != non_blocking:
    flags ^= os.O_NONBLOCK
  fcntl.fcntl(f.fileno(), fcntl.F_SETFL, flags) 
开发者ID:google,项目名称:macops,代码行数:13,代码来源:gmacpyutil.py


示例19: _set_cloexec

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def _set_cloexec(fd):
        try:
            flags = _fcntl.fcntl(fd, _fcntl.F_GETFD, 0)
        except OSError:
            pass
        else:
            # flags read successfully, modify
            flags |= _fcntl.FD_CLOEXEC
            _fcntl.fcntl(fd, _fcntl.F_SETFD, flags) 
开发者ID:war-and-code,项目名称:jawfish,代码行数:11,代码来源:tempfile.py


示例20: __init__

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def __init__(self, addr, requestHandler=SimpleXMLRPCRequestHandler,
                 logRequests=True, allow_none=False, encoding=None,
                 bind_and_activate=True, use_builtin_types=False):
        self.logRequests = logRequests
        SimpleXMLRPCDispatcher.__init__(self, allow_none, encoding, use_builtin_types)
        socketserver.TCPServer.__init__(self, addr, requestHandler, bind_and_activate)
        # [Bug #1222790] If possible, set close-on-exec flag; if a
        # method spawns a subprocess, the subprocess shouldn't have
        # the listening socket open.
        if fcntl is not None and hasattr(fcntl, 'FD_CLOEXEC'):
            flags = fcntl.fcntl(self.fileno(), fcntl.F_GETFD)
            flags |= fcntl.FD_CLOEXEC
            fcntl.fcntl(self.fileno(), fcntl.F_SETFD, flags) 
开发者ID:Soft8Soft,项目名称:verge3d-blender-addon,代码行数:17,代码来源:server.py


示例21: set_close_exec

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def set_close_exec(fd):
    flags = fcntl.fcntl(fd, fcntl.F_GETFD)
    fcntl.fcntl(fd, fcntl.F_SETFD, flags | fcntl.FD_CLOEXEC) 
开发者ID:tao12345666333,项目名称:tornado-zh,代码行数:5,代码来源:posix.py


示例22: _set_nonblocking

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def _set_nonblocking(fd):
    flags = fcntl.fcntl(fd, fcntl.F_GETFL)
    fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) 
开发者ID:tao12345666333,项目名称:tornado-zh,代码行数:5,代码来源:posix.py


示例23: _set_nonblocking

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def _set_nonblocking(self, fd):
        flags = fcntl.fcntl(fd, fcntl.F_GETFL)
        fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) 
开发者ID:tao12345666333,项目名称:tornado-zh,代码行数:5,代码来源:twisted_test.py


示例24: activate

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def activate(self):
        """Set up signal handlers.
        On Windows this uses a QTimer to periodically hand control over to
        Python so it can handle signals.
        On Unix, it uses a QSocketNotifier with os.set_wakeup_fd to get
        notified.
        """
        self._orig_handlers[signal.SIGINT] = signal.signal(
            signal.SIGINT, self.interrupt)
        self._orig_handlers[signal.SIGTERM] = signal.signal(
            signal.SIGTERM, self.interrupt)
        if utils.is_posix and hasattr(signal, 'set_wakeup_fd'):
            # pylint: disable=import-error,no-member,useless-suppression
            import fcntl
            read_fd, write_fd = os.pipe()
            for fd in [read_fd, write_fd]:
                flags = fcntl.fcntl(fd, fcntl.F_GETFL)
                fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)
            self._notifier = QSocketNotifier(typing.cast(sip.voidptr, read_fd),
                                             QSocketNotifier.Read,
                                             self)
            self._notifier.activated.connect(  # type: ignore[attr-defined]
                self.handle_signal_wakeup)
            self._orig_wakeup_fd = signal.set_wakeup_fd(write_fd)
            # pylint: enable=import-error,no-member,useless-suppression
        else:
            self._timer.start(1000)
            self._timer.timeout.connect(lambda: None)
        self._activated = True 
开发者ID:qutebrowser,项目名称:qutebrowser,代码行数:34,代码来源:crashsignal.py


示例25: set_binary_mode

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def set_binary_mode(f):
            try:
                fileno = f.fileno()
            except Exception:
                pass
            else:
                flags = fcntl.fcntl(fileno, fcntl.F_GETFL)
                fcntl.fcntl(fileno, fcntl.F_SETFL, flags & ~os.O_NONBLOCK)
            return f 
开发者ID:Frank-qlu,项目名称:recruit,代码行数:11,代码来源:_compat.py


示例26: _proper_fsync

# 需要导入模块: import fcntl [as 别名]
# 或者: from fcntl import fcntl [as 别名]
def _proper_fsync(fd):
            # https://lists.apple.com/archives/darwin-dev/2005/Feb/msg00072.html
            # https://developer.apple.com/library/mac/documentation/Darwin/Reference/ManPages/man2/fsync.2.html
            # https://github.com/untitaker/python-atomicwrites/issues/6
            fcntl.fcntl(fd, fcntl.F_FULLFSYNC) 
开发者ID:untitaker,项目名称:python-atomicwrites,代码行数:7,代码来源:__init__.py



注:本文中的fcntl.fcntl方法示例整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。