我们从Python开源项目中，提取了以下26个代码示例，用于说明如何使用mmap.PAGESIZE。
def _malloc(self, size): # returns a large enough block -- it might be much larger i = bisect.bisect_left(self._lengths, size) if i == len(self._lengths): length = self._roundup(max(self._size, size), mmap.PAGESIZE) self._size *= 2 info('allocating a new mmap of length %d', length) arena = Arena(length) self._arenas.append(arena) return (arena, 0, length) else: length = self._lengths[i] seq = self._len_to_seq[length] block = seq.pop() if not seq: del self._len_to_seq[length], self._lengths[i] (arena, start, stop) = block del self._start_to_block[(arena, start)] del self._stop_to_block[(arena, stop)] return block
def __init__(self, *args, **kwds): ## Create shared memory for rendered image #pg.dbg(namespace={'r': self}) if sys.platform.startswith('win'): self.shmtag = "pyqtgraph_shmem_" + ''.join([chr((random.getrandbits(20)%25) + 97) for i in range(20)]) self.shm = mmap.mmap(-1, mmap.PAGESIZE, self.shmtag) # use anonymous mmap on windows else: self.shmFile = tempfile.NamedTemporaryFile(prefix='pyqtgraph_shmem_') self.shmFile.write(b'\x00' * (mmap.PAGESIZE+1)) fd = self.shmFile.fileno() self.shm = mmap.mmap(fd, mmap.PAGESIZE, mmap.MAP_SHARED, mmap.PROT_WRITE) atexit.register(self.close) GraphicsView.__init__(self, *args, **kwds) self.scene().changed.connect(self.update) self.img = None self.renderTimer = QtCore.QTimer() self.renderTimer.timeout.connect(self.renderView) self.renderTimer.start(16)
def _malloc(self, size): # returns a large enough block -- it might be much larger i = bisect.bisect_left(self._lengths, size) if i == len(self._lengths): length = self._roundup(max(self._size, size), mmap.PAGESIZE) self._size *= 2 util.info('allocating a new mmap of length %d', length) arena = Arena(length) self._arenas.append(arena) return (arena, 0, length) else: length = self._lengths[i] seq = self._len_to_seq[length] block = seq.pop() if not seq: del self._len_to_seq[length], self._lengths[i] (arena, start, stop) = block del self._start_to_block[(arena, start)] del self._stop_to_block[(arena, stop)] return block
def __init__(self, ha_api_url, ble_adapter_name, mac_address, components): super().__init__() self.components = [] self.active_component = None component_instances = get_component_instances(components, mac_address) self.set_components(component_instances) self.manager = None self.ble_adapter_name = ble_adapter_name self.controller = None self.mac_address = mac_address self.battery_level = None # memory map using mmap to store nuimo battery level fd = os.open('/tmp/' + self.mac_address.replace(':', '-'), os.O_CREAT | os.O_TRUNC | os.O_RDWR) assert os.write(fd, b'\x00' * mmap.PAGESIZE) == mmap.PAGESIZE buf = mmap.mmap(fd, mmap.PAGESIZE, mmap.MAP_SHARED, mmap.PROT_WRITE) self.bl = ctypes.c_int.from_buffer(buf) self.bl.value = 0 offset = struct.calcsize(self.bl._type_) assert buf[offset] == 0
def __init__(self, path, page_count, read_only=False): # XXX TODO NOTE remove this line when write functionality is added. read_only = True # getting 'too many files open' error? increase the constant on the next line # (must be an exponent of 2) self._page_size = page_count * mmap.PAGESIZE # make sure we're sane here - allocation granularity needs to divide into page size! assert (self._page_size % mmap.ALLOCATIONGRANULARITY) == 0, 'page size is not a multiple of allocation granularity!' self._file = open(path, 'r+b') self._pages = dict() self.read_only = read_only self._path = path self.cursor = 0 super(MappedFile, self).__init__(self, base_offset=0, size=len(self))
def __init__(self, size=mmap.PAGESIZE): self._lastpid = os.getpid() self._lock = threading.Lock() self._size = size self._lengths = [] self._len_to_seq = {} self._start_to_block = {} self._stop_to_block = {} self._allocated_blocks = set() self._arenas = []
def __init__(self, size=mmap.PAGESIZE): self._lastpid = os.getpid() self._lock = threading.Lock() self._size = size self._lengths = [] self._len_to_seq = {} self._start_to_block = {} self._stop_to_block = {} self._allocated_blocks = set() self._arenas = [] # list of pending blocks to free - see free() comment below self._pending_free_blocks = []
def setUp(self): ''' Sets required params for dd workload and mounts the tmpfs ''' # Get required mem info self.mem_path = os.path.join(data_dir.get_tmp_dir(), 'thp_space') self.block_size = int(mmap.PAGESIZE) / 1024 # add mount point os.mkdir(self.mem_path) self.device = Partition(device="none", mountpoint=self.mem_path) self.device.mount(mountpoint=self.mem_path, fstype="tmpfs") free_space = (disk.freespace(self.mem_path)) / 1024 # Leaving out some free space in tmpfs self.count = (free_space / self.block_size) - 3
def get_nuimo_battery_level(mac_address): # pragma: no cover, try: fd = os.open('/tmp/' + mac_address.replace(':', '-'), os.O_RDONLY) except FileNotFoundError: logger.error("No memory mapped file named %s, return None battery level", mac_address.replace(':', '-')) return None buf = mmap.mmap(fd, mmap.PAGESIZE, mmap.MAP_SHARED, mmap.PROT_READ) result, = struct.unpack('i', buf[:4]) return result
