gistfile1.txt

import os
import mmap
from contextlib import contextmanager

BLOCK_SIZE = 4 * 1024

@contextmanager
def dopen(filename, blockno=1):
    try:
        fd = os.open(filename, os.O_CREAT | os.O_DIRECT | os.O_RDWR)
        os.ftruncate(fd, blockno * BLOCK_SIZE)
        yield fd
    finally:
        os.close(fd)

class DirectDevice(object):

    def __init__(self, filename, size=BLOCK_SIZE):
        self.filename = filename
        self.size = size

    def __enter__(self):
        flags = os.O_CREAT | os.O_DIRECT | os.O_RDWR
        self.fd = os.open(self.filename, flags)
        os.ftruncate(self.fd, self.size)
        return self

    def __exit__(self, *args, **kwargs):
        os.close(self.fd)

    def write(self, buff, size=None, offset=0):
        """Write size bytes from buff to offset
        """
        chunkno, start = divmod(offset, mmap.PAGESIZE)
        if not size:
            size = len(buff)
        m = mmap.mmap(self.fd, start + size, offset=chunkno * mmap.PAGESIZE)
        m[start:start + size] = buff[:size]

    def read(self, size=mmap.PAGESIZE, offset=0):
        """Read size bytes start at offset
        """
        chunkno, start = divmod(offset, mmap.PAGESIZE)
        end = start + size
        return mmap.mmap(self.fd, end,
                         offset=chunkno * mmap.PAGESIZE)[start:end]