added async_write option to data_helper

added async_write option to data_helper Change-Id: Ia844d816de893b76f7289698fc47567aef6128fe
couchbase · Jun 5, 2011 · 27b7c7e · 27b7c7e
1 parent c0d2a86
commit 27b7c7e
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 6 deletions.
diff --git a/lib/memcached/helper/data_helper.py b/lib/memcached/helper/data_helper.py
@@ -31,7 +31,8 @@ def create_threads(servers=None,
                        number_of_threads=50,
                        override_vBucketId=-1,
                        write_only=False,
-                       moxi=True):
+                       moxi=True,
+                       async_write=True):
         log = logger.Logger.get_logger()
         if not servers:
             raise MemcachedClientHelperExcetion(errorcode='invalid_argument',
@@ -80,7 +81,8 @@ def create_threads(servers=None,
                                   ignore_how_many_errors=5000,
                                   override_vBucketId=override_vBucketId,
                                   write_only=write_only,
-                                  moxi=moxi)
+                                  moxi=moxi,
+                                  async_write=async_write)
             threads.append(thread)
 
         return threads
@@ -397,7 +399,8 @@ def __init__(self,
                  override_vBucketId=-1,
                  terminate_in_minutes=60,
                  write_only=False,
-                 moxi=True):
+                 moxi=True,
+                 async_write=False):
         threading.Thread.__init__(self)
         self.log = logger.Logger.get_logger()
         self.serverInfo = serverInfo
@@ -425,6 +428,7 @@ def __init__(self,
                      'baseuuid': self._base_uuid}
         self.write_only = write_only
         self.aborted = False
+        self.async_write = async_write
 
     def inserted_keys_count(self):
         return self._inserted_keys_count
@@ -505,7 +509,10 @@ def run(self):
             try:
                 if self.override_vBucketId >= 0:
                     client.vbucketId = self.override_vBucketId
-                client.set(key, 0, 0, selected['value'])
+                if self.async_write:
+                    client.send_set(key, 0, 0, selected['value'])
+                else:
+                    client.set(key, 0, 0, selected['value'])
                 self._inserted_keys_count += 1
                 backoff_count = 0
             except MemcachedError as error:
@@ -559,7 +566,10 @@ def run(self):
                 try:
                     if self.override_vBucketId >= 0:
                         client.vbucketId = self.override_vBucketId
-                    client.set(key, 0, 0, key)
+                    if self.async_write:
+                        client.send_set(key, 0, 0, selected['value'])
+                    else:
+                        client.set(key, 0, 0, selected['value'])
                     self._inserted_keys_count += 1
                 except MemcachedError:
                     self._rejected_keys_count += 1

diff --git a/pytests/combotests.py b/pytests/combotests.py
@@ -107,7 +107,8 @@ def common_test_body(self, replica, load_ratio, timeout=10):
                                                        number_of_threads=20,
                                                        number_of_items=4000000000,
                                                        moxi=False,
-                                                       write_only=True)
+                                                       write_only=True,
+                                                       async_write=True)
         for thread in threads:
             thread.start()
         while time.time() < ( start_time + 60 * timeout):