Merge branch 'device'

Author: Byron

smmap/buf.py

@@ -0,0 +1,108 @@
+"""Module with a simple buffer implementation using the memory manager"""
+from mman import MemoryCursor
+
+import sys
+
+__all__ = ["MappedMemoryBuffer"]
+
+class MappedMemoryBuffer(object):
+	"""A buffer like object which allows direct byte-wise object and slicing into 
+	memory of a mapped file. The mapping is controlled by an underlying memory manager.
+	
+	A buffer, once initialized, stays put on providing access to eactly one path.
+	A custom interface allows you to change paths mid way, and to optimize 
+	the resource usage.
+	
+	Please note that this type is only fully usable if you configure it with the 
+	MappedMemoryManager to use.
+	
+	The buffer is relative, that is if you map an offset, index 0 will map to the 
+	first byte at your given offset."""
+	__slots__ = '_c'		# our cursor
+	
+	#{ Configuration
+	# A subclass must provide an instance of a (usually global) MappedMemoryManager
+	manager = None
+	#}END configuration
+	
+	def __init__(self, path = None, offset = 0, size = sys.maxint, flags = 0):
+		"""Initalize the instance to operate on the given path if given.
+		:param path: if not None, the path to the file you want to access
+			If None, you have call begin_access before using the buffer 
+		:param offset: absolute offset in bytes
+		:param size: the total size of the mapping. Defaults to the maximum possible size
+		:param flags: Additional flags to be passed to os.open
+		:raise ValueError: if the buffer could not achieve a valid state"""
+		self._c = MemoryCursor(self.manager)
+		assert self.manager is not None, "Require the cls.manager variable to be set in subclass"
+		if path and not self.begin_access(path, offset, size, flags):
+			raise ValueError("Failed to allocate the buffer - probably the given offset is out of bounds")
+		# END handle offset
+
+	def __del__(self):
+		self.end_access()
+		
+	def __getitem__(self, i):
+		c = self._c
+		assert c.is_valid()
+		if not c.includes_ofs(i):
+			c.use_region(i, 1)
+		# END handle region usage
+		return c.buffer()[i-c.ofs_begin()]
+	
+	def __getslice__(self, i, j):
+		c = self._c
+		# fast path, slice fully included - safes a concatenate operation and 
+		# should be the default
+		assert c.is_valid()
+		if (c.ofs_begin() <= i) and (j < c.ofs_end()):
+			b = c.ofs_begin()
+			return c.buffer()[i-b:j-b]
+		else:
+			l = j-i					# total length
+			ofs = i
+			# Keeping tokens in a list could possible be faster, but the list
+			# overhead outweighs the benefits (tested) !
+			md = str()
+			while l:
+				c.use_region(ofs, l)
+				d = c.buffer()[:l]
+				ofs += len(d)
+				l -= len(d)
+				md += d
+			#END while there are bytes to read
+			return md
+		# END fast or slow path
+	#{ Interface
+	
+	def begin_access(self, path = None, offset = 0, size = sys.maxint, flags = 0):
+		"""Call this before the first use of this instance. The method was already
+		called by the constructor in case sufficient information was provided.
+		
+		For more information no the parameters, see the __init__ method
+		:param path: if path is empty or None the existing path will be used if possible. 
+		:return: True if the buffer can be used"""
+		if path and (not self._c.is_associated() or self._c.path() != path):
+			self._c = self.manager.make_cursor(path)
+		#END get associated cursor
+		
+		# reuse existing cursors if possible
+		if self._c.is_associated():
+			return self._c.use_region(offset, size, flags).is_valid()
+		return False
+		
+	def end_access(self):
+		"""Call this method once you are done using the instance. It is automatically 
+		called on destruction, and should be called just in time to allow system
+		resources to be freed.
+		
+		Once you called end_access, you must call begin access before reusing this instance!"""
+		self._c.unuse_region()
+		
+	def cursor(self):
+		""":return: the currently set cursor which provides access to the data"""
+		return self._c
+		
+	#}END interface
+
+

smmap/mman.py

@@ -78,10 +78,12 @@ def assign(self, rhs):
 		self._destroy()
 		self._copy_from(rhs)
 
-	def use_region(self, offset, size, _is_recursive=False):
+	def use_region(self, offset, size, flags = 0, _is_recursive=False):
 		"""Assure we point to a window which allows access to the given offset into the file
 		:param offset: absolute offset in bytes into the file
 		:param size: amount of bytes to map
+		:param flags: additional flags to be given to os.open in case a file handle is initially opened
+			for mapping. Has no effect if a region can actually be reused.
 		:return: this instance - it should be queried for whether it points to a valid memory region.
 			This is not the case if the mapping failed becaues we reached the end of the file
 		:note: The size actually mapped may be smaller than the given size. If that is the case,
@@ -106,6 +108,8 @@ def use_region(self, offset, size, _is_recursive=False):
 				return self
 			# END handle offset too large
 
+			# bisect to find an existing region. The c++ implementation cannot 
+			# do that as it uses a linked list for regions.
 			existing_region = None
 			a = self._rlist
 			lo = 0
@@ -181,7 +185,7 @@ def use_region(self, offset, size, _is_recursive=False):
 					if man._handle_count >= man._max_handle_count:
 						raise Exception
 					#END assert own imposed max file handles
-					self._region = MappedRegion(a.path(), mid.ofs, mid.size)
+					self._region = MappedRegion(a.path(), mid.ofs, mid.size, flags)
 				except Exception:
 					# apparently we are out of system resources or hit a limit
 					# As many more operations are likely to fail in that condition (
@@ -195,7 +199,7 @@ def use_region(self, offset, size, _is_recursive=False):
 						raise
 					#END handle existing recursion
 					man._collect_lru_region(0)
-					return self.use_region(offset, size, True) 
+					return self.use_region(offset, size, flags, True) 
 				#END handle exceptions
 
 				man._handle_count += 1
@@ -218,11 +222,15 @@ def unuse_region(self):
 			to unuse the region once you are done reading from it in persistent cursors as it 
 			helps to free up resource more quickly"""
 		self._region = None
+		# note: should reset ofs and size, but we spare that for performance. Its not 
+		# allowed to query information if we are not valid !
 
 	def buffer(self):
 		"""Return a buffer object which allows access to our memory region from our offset
 		to the window size. Please note that it might be smaller than you requested
-		:note: You can only obtain a buffer if this instance is_valid() !"""
+		:note: You can only obtain a buffer if this instance is_valid() !
+		:note: buffers should not be cached passed the duration of your access as it will 
+			prevent resources from being freed even though they might not be accounted for anymore !"""
 		return buffer(self._region.buffer(), self._ofs, self._size)
 
 	def is_valid(self):
@@ -234,7 +242,8 @@ def is_associated(self):
 		return self._rlist is not None
 
 	def ofs_begin(self):
-		""":return: offset to the first byte pointed to by our cursor"""
+		""":return: offset to the first byte pointed to by our cursor
+		:note: only if is_valid() is True"""
 		return self._region._b + self._ofs
 
 	def ofs_end(self):
@@ -332,6 +341,7 @@ def _collect_lru_region(self, size):
 		:param size: size of the region we want to map next (assuming its not already mapped partially or full
 			if 0, we try to free any available region
 		:raise RegionCollectionError:
+		:return: Amount of freed regions
 		:todo: implement a case where all unusued regions are discarded efficiently. Currently its only brute force"""
 		num_found = 0
 		while (size == 0) or (self._memory_size + size > self._max_memory_size):
@@ -361,6 +371,8 @@ def _collect_lru_region(self, size):
 			self._handle_count -= 1
 		#END while there is more memory to free
 
+		return num_found
+		
 	#{ Interface 
 	def make_cursor(self, path):
 		""":return: a cursor pointing to the given path. It can be used to map new regions of the file into memory"""
@@ -371,6 +383,11 @@ def make_cursor(self, path):
 		# END obtain region for path
 		return MemoryCursor(self, regions)
 
+	def collect(self):
+		"""Collect all available free-to-collect mapped regions
+		:return: Amount of freed handles"""
+		return self._collect_lru_region(0)
+		
 	def num_file_handles(self):
 		""":return: amount of file handles in use. Each mapped region uses one file handle"""
 		return self._handle_count

smmap/stream.py

@@ -0,0 +1,102 @@
+from lib import TestBase, FileCreator
+
+from smmap.mman import MappedMemoryManager
+from smmap.buf import *
+
+from random import randint
+from time import time
+import sys
+
+class TestBuffer(MappedMemoryBuffer):
+	#{ Configuration
+	manager = MappedMemoryManager()
+	#} END configuration
+	
+
+class TestBuf(TestBase):
+	
+	def test_basics(self):
+		self.failUnlessRaises(AssertionError, MappedMemoryBuffer)			# needs subclass
+		fc = FileCreator(self.k_window_test_size, "buffer_test")
+		
+		# invalid paths fail upon construction
+		self.failUnlessRaises(OSError, TestBuffer, "somefile")				# invalid file
+		self.failUnlessRaises(ValueError, TestBuffer, fc.path, fc.size)		# offset too large
+		
+		buf = TestBuffer()												# can create uninitailized buffers
+		assert not buf.cursor().is_valid() and not buf.cursor().is_associated()
+		
+		# can call end access any time
+		buf.end_access()
+		buf.end_access()
+		
+		# begin access can revive it, if the offset is suitable
+		offset = 100
+		assert buf.begin_access(fc.path, fc.size) == False
+		assert buf.begin_access(fc.path, offset) == True
+		
+		# empty begin access keeps it valid on the same path, but alters the offset
+		assert buf.begin_access() == True
+		assert buf.cursor().is_valid()
+		
+		# simple access
+		data = open(fc.path, 'rb').read()
+		assert data[offset] == buf[0]
+		assert data[offset:offset*2] == buf[0:offset]
+		
+		# end access makes its cursor invalid
+		buf.end_access()
+		assert not buf.cursor().is_valid()
+		assert buf.cursor().is_associated()			# but it remains associated
+		
+		# an empty begin access fixes it up again
+		assert buf.begin_access() == True and buf.cursor().is_valid()
+		del(buf)		# ends access automatically
+		
+		man = TestBuffer.manager
+		assert man.num_file_handles() == 1
+		
+		# PERFORMANCE
+		# blast away with rnadom access and a full mapping - we don't want to 
+		# exagerate the manager's overhead, but measure the buffer overhead
+		# We do it once with an optimal setting, and with a worse manager which 
+		# will produce small mappings only !
+		max_num_accesses = 1000
+		for manager, man_id in ( (man, 'optimal'), 
+								(MappedMemoryManager(window_size=fc.size/100, max_memory_size=fc.size/3, max_open_handles=15), 'worst case')):
+			TestBuffer.manager = manager
+			buf = TestBuffer(fc.path)
+			assert manager.num_file_handles() == 1
+			for access_mode in range(2):	# single, multi
+				num_accesses_left = max_num_accesses
+				num_bytes = 0
+				fsize = fc.size
+				
+				st = time()
+				buf.begin_access()
+				while num_accesses_left:
+					num_accesses_left -= 1
+					if access_mode:	# multi
+						ofs_start = randint(0, fsize)
+						ofs_end = randint(ofs_start, fsize)
+						d = buf[ofs_start:ofs_end]
+						assert len(d) == ofs_end - ofs_start
+						assert d == data[ofs_start:ofs_end]
+						num_bytes += len(d)
+					else:
+						pos = randint(0, fsize)
+						assert buf[pos] == data[pos]
+						num_bytes += 1
+					#END handle mode
+				# END handle num accesses
+				
+				buf.end_access()
+				assert manager.num_file_handles()
+				assert manager.collect()
+				assert manager.num_file_handles() == 0
+				elapsed  = time() - st
+				mb = float(1000*1000)
+				mode_str = (access_mode and "slice") or "single byte"
+				sys.stderr.write("%s: Made %i random %s accesses to buffer reading a total of %f mb in %f s (%f mb/s)\n" % (man_id, max_num_accesses, mode_str, num_bytes/mb, elapsed, (num_bytes/mb)/elapsed)) 
+			# END handle access mode
+		# END for each manager

smmap/test/test_buf.py

@@ -75,7 +75,8 @@ def test_memman_operation(self):
 		assert len(data) == fc.size
 
 		# small windows, a reasonable max memory. Not too many regions at once
-		man = MappedMemoryManager(fc.size / 100, fc.size / 3, 15)
+		max_num_handles = 15
+		man = MappedMemoryManager(window_size=fc.size / 100, max_memory_size=fc.size / 3, max_open_handles=max_num_handles)
 		c = man.make_cursor(fc.path)
 
 		# still empty (more about that is tested in test_memory_manager()
@@ -129,7 +130,7 @@ def test_memman_operation(self):
 
 		# iterate through the windows, verify data contents
 		# this will trigger map collection after a while
-		max_random_accesses = 15000
+		max_random_accesses = 5000
 		num_random_accesses = max_random_accesses
 		memory_read = 0
 		st = time()
@@ -157,9 +158,14 @@ def test_memman_operation(self):
 			assert not includes_ofs(base_offset+csize)
 		# END while we should do an access
 		elapsed = time() - st
-		mb = 1000 * 1000
+		mb = float(1000 * 1000)
 		sys.stderr.write("Read %i mb of memory with %i random accesses in %fs (%f mb/s)\n" 
 						% (memory_read/mb, max_random_accesses, elapsed, (memory_read/mb)/elapsed))
-		
+
 		# an offset as large as the size doesn't work !
-		assert not c.use_region(fc.size, size).is_valid() 
+		assert not c.use_region(fc.size, size).is_valid()
+		
+		# collection - it should be able to collect all
+		assert man.num_file_handles()
+		assert man.collect()
+		assert man.num_file_handles() == 0

smmap/test/test_mman.py

@@ -93,18 +93,19 @@ class MappedRegion(object):
 	#END handle additional slot
 
 
-	def __init__(self, path, ofs, size):
+	def __init__(self, path, ofs, size, flags = 0):
 		"""Initialize a region, allocate the memory map
 		:param path: path to the file to map
 		:param ofs: **aligned** offset into the file to be mapped 
 		:param size: if size is larger then the file on disk, the whole file will be
 			allocated the the size automatically adjusted
+		:param flags: additional flags to be given when opening the file. 
 		:raise Exception: if no memory can be allocated"""
 		self._b = ofs
 		self._size = 0
 		self._uc = 0
 
-		fd = os.open(path, os.O_RDONLY|getattr(os, 'O_BINARY', 0))
+		fd = os.open(path, os.O_RDONLY|getattr(os, 'O_BINARY', 0)|flags)
 		try:
 			kwargs = dict(access=ACCESS_READ, offset=ofs)
 			corrected_size = size