Skip to content

Instantly share code, notes, and snippets.

@schmichael
Created September 6, 2011 05:44
Show Gist options
  • Save schmichael/1196686 to your computer and use it in GitHub Desktop.
Save schmichael/1196686 to your computer and use it in GitHub Desktop.
WIP Convenience wrapper around Python's mmap.mmap class
"""Public domain"""
import mmap
import os
PAGESIZE = mmap.PAGESIZE
def fileno_to_filename(fd):
"""Attempt to resolve fd to a filename or return None"""
try:
#XXX This should work on Linux... any ideas for other platforms?
filename = os.readlink('/proc/%d/fd/%d' % (os.getpid(), fd))
except OSError:
filename = None
return filename
class MemoryMappedFile(mmap.mmap):
def __new__(cls, file_, shared=True, write=True, initial_size=None):
"""
Initialize a new memory mapped file instance.
`file_` is a filename or file-like object (has a fileno attribute)
`shared` means other processes can access this mmap
Setting `write` to False makes access read only
`initial_size` will default to either the size of file_ or 1 PAGESIZE
"""
# file_ => fileno
if isinstance(file_, int):
fileno = file_
filename = fileno_to_filename(fileno)
elif hasattr(file_, 'fileno'):
fileno = file_.fileno
if hasattr(file_, 'filename'):
filename = file_.filename
else:
filename = fileno_to_filename(fileno)
else:
filename = file_
if write:
fileno = os.open(file_, os.O_CREAT | os.O_TRUNC | os.O_RDWR)
else:
fileno = os.open(file_, os.O_READ)
# initial_size
initial_size = PAGESIZE
# initialize file 1 PAGESIZE at a time if writable
if write:
for _ in range(0, initial_size, PAGESIZE):
os.write(fileno, '\x00' * PAGESIZE)
# access
access = mmap.MAP_SHARED if shared else mmap.MAP_PRIVATE
# protection
prot = mmap.PROT_WRITE if write else mmap.PROT_READ
inst = mmap.mmap.__new__(cls, fileno, initial_size, access, prot)
inst._filename = filename
inst._fileno = fileno
inst._shared = shared
inst._writable = write
return inst
@property
def filename(self):
"""May be None if mmap created with an open fd"""
return self._filename
@property
def fileno(self):
return self._fileno
@property
def private(self):
return not self._shared
@property
def shared(self):
return self._shared
@property
def writable(self):
return self._writable
def __contains__(self, item):
# Using find is far more efficient than letting "in" use the old
# iterator protocol
return bool(self.find(item))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment