Implemented all async methods, including test which shows how to chain the async method together

Author: Byron

__init__.py

@@ -1,5 +1,17 @@
 """Initialize the object database module"""
 
+import sys
+import os
+
+#{ Initialization
+def _init_externals():
+	"""Initialize external projects by putting them into the path"""
+	sys.path.append(os.path.join(os.path.dirname(__file__), 'ext'))
+	
+#} END initialization
+
+_init_externals()
+
 # default imports
 from db import *
 from stream import *

db.py

@@ -14,6 +14,7 @@
 	)
 
 from util import (
+		pool,
 		ENOENT,
 		to_hex_sha,
 		exists,
@@ -32,6 +33,11 @@
 	stream_copy
 	)
 
+
+from async import (
+		ChannelThreadTask
+	)
+
 import tempfile
 import mmap
 import os
@@ -40,6 +46,7 @@
 __all__ = ('ObjectDBR', 'ObjectDBW', 'FileDBBase', 'LooseObjectDB', 'PackedDB', 
 			'CompoundDB', 'ReferenceDB', 'GitObjectDB' )
 
+
 class ObjectDBR(object):
 	"""Defines an interface for object database lookup.
 	Objects are identified either by hex-sha (40 bytes) or 
@@ -52,34 +59,48 @@ def __contains__(self, sha):
 	def has_object(self, sha):
 		"""
 		:return: True if the object identified by the given 40 byte hexsha or 20 bytes
-			binary sha is contained in the database
-		:raise BadObject:"""
+			binary sha is contained in the database"""
 		raise NotImplementedError("To be implemented in subclass")
 
+	def has_object_async(self, reader):
+		"""Return a reader yielding information about the membership of objects
+		as identified by shas
+		:param reader: Reader yielding 20 byte or 40 byte shas.
+		:return: async.Reader yielding tuples of (sha, bool) pairs which indicate
+			whether the given sha exists in the database or not"""
+		task = ChannelThreadTask(reader, str(self.has_object_async), lambda sha: (sha, self.has_object(sha)))
+		return pool.add_task(task) 
+		
 	def info(self, sha):
 		""" :return: OInfo instance
 		:param sha: 40 bytes hexsha or 20 bytes binary sha
 		:raise BadObject:"""
 		raise NotImplementedError("To be implemented in subclass")
 
-	def info_async(self, input_channel):
+	def info_async(self, reader):
 		"""Retrieve information of a multitude of objects asynchronously
-		:param input_channel: Channel yielding the sha's of the objects of interest
-		:return: Channel yielding OInfo|InvalidOInfo, in any order"""
-		raise NotImplementedError("To be implemented in subclass")
+		:param reader: Channel yielding the sha's of the objects of interest
+		:return: async.Reader yielding OInfo|InvalidOInfo, in any order"""
+		task = ChannelThreadTask(reader, str(self.info_async), self.info)
+		return pool.add_task(task)
 
 	def stream(self, sha):
 		""":return: OStream instance
 		:param sha: 40 bytes hexsha or 20 bytes binary sha
 		:raise BadObject:"""
 		raise NotImplementedError("To be implemented in subclass")
 
-	def stream_async(self, input_channel):
+	def stream_async(self, reader):
 		"""Retrieve the OStream of multiple objects
-		:param input_channel: see ``info``
+		:param reader: see ``info``
 		:param max_threads: see ``ObjectDBW.store``
-		:return: Channel yielding OStream|InvalidOStream instances in any order"""
-		raise NotImplementedError("To be implemented in subclass")
+		:return: async.Reader yielding OStream|InvalidOStream instances in any order
+		:note: depending on the system configuration, it might not be possible to 
+			read all OStreams at once. Instead, read them individually using reader.read(x)
+			where x is small enough."""
+		# base implementation just uses the stream method repeatedly
+		task = ChannelThreadTask(reader, str(self.stream_async), self.stream)
+		return pool.add_task(task)
 
 	#} END query interface
 
@@ -114,21 +135,23 @@ def store(self, istream):
 		:raise IOError: if data could not be written"""
 		raise NotImplementedError("To be implemented in subclass")
 
-	def store_async(self, input_channel):
+	def store_async(self, reader):
 		"""Create multiple new objects in the database asynchronously. The method will 
 		return right away, returning an output channel which receives the results as 
 		they are computed.
 		
 		:return: Channel yielding your IStream which served as input, in any order.
 			The IStreams sha will be set to the sha it received during the process, 
 			or its error attribute will be set to the exception informing about the error.
-		:param input_channel: Channel yielding IStream instance.
-			As the same instances will be used in the output channel, you can create a map
-			between the id(istream) -> istream
-		:note:As some ODB implementations implement this operation as atomic, they might 
+		:param reader: async.Reader yielding IStream instances.
+			The same instances will be used in the output channel as were received
+			in by the Reader.
+		:note:As some ODB implementations implement this operation atomic, they might 
 			abort the whole operation if one item could not be processed. Hence check how 
 			many items have actually been produced."""
-		raise NotImplementedError("To be implemented in subclass")
+		# base implementation uses store to perform the work
+		task = ChannelThreadTask(reader, str(self.store_async), self.store) 
+		return pool.add_task(task)
 
 	#} END edit interface
 

ext/async

@@ -1 +1,12 @@
 
+import gitdb.util
+
+#{ Initialization 
+def _init_pool():
+	"""Assure the pool is actually threaded"""
+	size = 2
+	print "Setting ThreadPool to %i" % size
+	gitdb.util.pool.set_size(size)
+
+
+#} END initialization

test/__init__.py

@@ -30,7 +30,7 @@ def with_rw_directory(func):
 	"""Create a temporary directory which can be written to, remove it if the 
 	test suceeds, but leave it otherwise to aid additional debugging"""
 	def wrapper(self):
-		path = tempfile.mktemp(suffix=func.__name__)
+		path = tempfile.mktemp(prefix=func.__name__)
 		os.mkdir(path)
 		try:
 			return func(self, path)

test/lib.py

@@ -10,6 +10,8 @@
 from gitdb.exc import BadObject
 from gitdb.typ import str_blob_type
 
+from async import IteratorReader
+
 from cStringIO import StringIO
 import os
 
@@ -78,15 +80,99 @@ def _assert_object_writing(self, db):
 					new_ostream = db.ostream()
 
 					# note: only works as long our store write uses the same compression
-					# level, which is zip
+					# level, which is zip_best
 					assert ostream.getvalue() == new_ostream.getvalue()
 			# END for each data set
 		# END for each dry_run mode
+		
+	def _assert_object_writing_async(self, db):
+		"""Test generic object writing using asynchronous access"""
+		ni = 5000
+		def istream_generator(offset=0, ni=ni):
+			for data_src in xrange(ni):
+				data = str(data_src + offset)
+				yield IStream(str_blob_type, len(data), StringIO(data))
+			# END for each item
+		# END generator utility
+		
+		# for now, we are very trusty here as we expect it to work if it worked
+		# in the single-stream case
+		
+		# write objects
+		reader = IteratorReader(istream_generator())
+		istream_reader = db.store_async(reader)
+		istreams = istream_reader.read()		# read all
+		assert istream_reader.task().error() is None
+		assert len(istreams) == ni
+		
+		for stream in istreams:
+			assert stream.error is None
+			assert len(stream.sha) == 40
+			assert isinstance(stream, IStream)
+		# END assert each stream
+		
+		# test has-object-async - we must have all previously added ones
+		reader = IteratorReader( istream.sha for istream in istreams )
+		hasobject_reader = db.has_object_async(reader)
+		count = 0
+		for sha, has_object in hasobject_reader:
+			assert has_object
+			count += 1
+		# END for each sha
+		assert count == ni
+		
+		# read the objects we have just written
+		reader = IteratorReader( istream.sha for istream in istreams )
+		ostream_reader = db.stream_async(reader)
+		
+		# read items individually to prevent hitting possible sys-limits
+		count = 0
+		for ostream in ostream_reader:
+			assert isinstance(ostream, OStream)
+			count += 1
+		# END for each ostream
+		assert ostream_reader.task().error() is None
+		assert count == ni
+		
+		# get info about our items
+		reader = IteratorReader( istream.sha for istream in istreams )
+		info_reader = db.info_async(reader)
+		
+		count = 0
+		for oinfo in info_reader:
+			assert isinstance(oinfo, OInfo)
+			count += 1
+		# END for each oinfo instance
+		assert count == ni
+		
+		  
+		# combined read-write using a converter
+		# add 2500 items, and obtain their output streams
+		nni = 2500
+		reader = IteratorReader(istream_generator(offset=ni, ni=nni))
+		istream_to_sha = lambda istreams: [ istream.sha for istream in istreams ]
+		
+		istream_reader = db.store_async(reader)
+		istream_reader.set_post_cb(istream_to_sha)
+		
+		ostream_reader = db.stream_async(istream_reader)
+		
+		count = 0
+		# read it individually, otherwise we might run into the ulimit
+		for ostream in ostream_reader:
+			assert isinstance(ostream, OStream)
+			count += 1
+		# END for each ostream
+		assert count == nni
+		
+		
+		
 
 	@with_rw_directory
 	def test_writing(self, path):
 		ldb = LooseObjectDB(path)
 
 		# write data
 		self._assert_object_writing(ldb)
+		self._assert_object_writing_async(ldb)
 

test/test_db.py

@@ -2,11 +2,21 @@
 import os
 import errno
 
+from async import ThreadPool
+
 try:
     import hashlib
 except ImportError:
     import sha
 
+#{ Globals
+
+# A pool distributing tasks, initially with zero threads, hence everything 
+# will be handled in the main thread
+pool = ThreadPool(0)
+
+#} END globals
+
 
 #{ Aliases