Merge branch 'master' of https://github.com/pinterest/pymemcachev1.2.9

4 files changed, 491 insertions, 245 deletions

diff --git a/README.md b/README.md
index d6caa39..6dfdb76 100644
--- a/README.md
+++ b/README.md
@@ -95,3 +95,4 @@ Credits
 * [Julien Danjou](http://github.com/jd)
 * [INADA Naoki](http://github.com/methane)
 * [James Socol](http://github.com/jsocol)
+* [Joshua Harlow](http://github.com/harlowja)
diff --git a/pymemcache/client.py b/pymemcache/client.py
index fd31589..18a6b4a 100644
--- a/pymemcache/client.py
+++ b/pymemcache/client.py
@@ -73,6 +73,8 @@ __author__ = "Charles Gordon"
 import socket
 import six
 
+from pymemcache import pool
+
 
 RECV_SIZE = 4096
 VALID_STORE_RESULTS = {
@@ -152,6 +154,23 @@ class MemcacheUnexpectedCloseError(MemcacheServerError):
     pass
 
 
+# Common helper functions.
+
+def _check_key(key, key_prefix=b''):
+    """Checks key and add key_prefix."""
+    if isinstance(key, six.text_type):
+        try:
+            key = key.encode('ascii')
+        except UnicodeEncodeError:
+            raise MemcacheIllegalInputError("No ascii key: %r" % (key,))
+    key = key_prefix + key
+    if b' ' in key:
+        raise MemcacheIllegalInputError("Key contains spaces: %r" % (key,))
+    if len(key) > 250:
+        raise MemcacheIllegalInputError("Key is too long: %r" % (key,))
+    return key
+
+
 class Client(object):
     """
     A client for a single memcached server.
@@ -266,7 +285,6 @@ class Client(object):
         self.ignore_exc = ignore_exc
         self.socket_module = socket_module
         self.sock = None
-        self.buf = b''
         if isinstance(key_prefix, six.text_type):
             key_prefix = key_prefix.encode('ascii')
         if not isinstance(key_prefix, bytes):
@@ -275,17 +293,7 @@ class Client(object):
 
     def check_key(self, key):
         """Checks key and add key_prefix."""
-        if isinstance(key, six.text_type):
-            try:
-                key = key.encode('ascii')
-            except UnicodeEncodeError as e:
-                raise MemcacheIllegalInputError("No ascii key: %r" % (key,))
-        key = self.key_prefix + key
-        if b' ' in key:
-            raise MemcacheIllegalInputError("Key contains spaces: %r" % (key,))
-        if len(key) > 250:
-            raise MemcacheIllegalInputError("Key is too long: %r" % (key,))
-        return key
+        return _check_key(key, key_prefix=self.key_prefix)
 
     def _connect(self):
         sock = self.socket_module.socket(self.socket_module.AF_INET,
@@ -307,7 +315,6 @@ class Client(object):
             except Exception:
                 pass
         self.sock = None
-        self.buf = b''
 
     def set(self, key, value, expire=0, noreply=True):
         """
@@ -694,9 +701,10 @@ class Client(object):
 
             self.sock.sendall(cmd)
 
+            buf = b''
             result = {}
             while True:
-                self.buf, line = _readline(self.sock, self.buf)
+                buf, line = _readline(self.sock, buf)
                 self._raise_errors(line, name)
 
                 if line == b'END':
@@ -711,9 +719,7 @@ class Client(object):
                             raise ValueError("Unable to parse line %s: %s"
                                              % (line, str(e)))
 
-                    self.buf, value = _readvalue(self.sock,
-                                                 self.buf,
-                                                 int(size))
+                    buf, value = _readvalue(self.sock, buf, int(size))
                     key = checked_keys[key]
 
                     if self.deserializer:
@@ -767,7 +773,8 @@ class Client(object):
             if noreply:
                 return True
 
-            self.buf, line = _readline(self.sock, self.buf)
+            buf = b''
+            buf, line = _readline(self.sock, buf)
             self._raise_errors(line, name)
 
             if line in VALID_STORE_RESULTS[name]:
@@ -816,6 +823,182 @@ class Client(object):
         self.delete(key, noreply=True)
 
 
+class PooledClient(object):
+    """A thread-safe pool of clients (with the same client api)."""
+
+    def __init__(self,
+                 server,
+                 serializer=None,
+                 deserializer=None,
+                 connect_timeout=None,
+                 timeout=None,
+                 no_delay=False,
+                 ignore_exc=False,
+                 socket_module=socket,
+                 key_prefix=b'',
+                 max_pool_size=None):
+        self.server = server
+        self.serializer = serializer
+        self.deserializer = deserializer
+        self.connect_timeout = connect_timeout
+        self.timeout = timeout
+        self.no_delay = no_delay
+        self.ignore_exc = ignore_exc
+        self.socket_module = socket_module
+        if isinstance(key_prefix, six.text_type):
+            key_prefix = key_prefix.encode('ascii')
+        if not isinstance(key_prefix, bytes):
+            raise TypeError("key_prefix should be bytes.")
+        self.key_prefix = key_prefix
+        self.client_pool = pool.ObjectPool(
+            self._create_client,
+            after_remove=lambda client: client.close(),
+            max_size=max_pool_size)
+
+    def check_key(self, key):
+        """Checks key and add key_prefix."""
+        return _check_key(key, key_prefix=self.key_prefix)
+
+    def _create_client(self):
+        client = Client(self.server,
+                        serializer=self.serializer,
+                        deserializer=self.deserializer,
+                        connect_timeout=self.connect_timeout,
+                        timeout=self.timeout,
+                        no_delay=self.no_delay,
+                        # We need to know when it fails *always* so that we
+                        # can remove/destroy it from the pool...
+                        ignore_exc=False,
+                        socket_module=self.socket_module,
+                        key_prefix=self.key_prefix)
+        return client
+
+    def close(self):
+        self.client_pool.clear()
+
+    def set(self, key, value, expire=0, noreply=True):
+        with self.client_pool.get_and_release(destroy_on_fail=True) as client:
+            return client.set(key, value, expire=expire, noreply=noreply)
+
+    def set_many(self, values, expire=0, noreply=True):
+        with self.client_pool.get_and_release(destroy_on_fail=True) as client:
+            return client.set_many(values, expire=expire, noreply=noreply)
+
+    def replace(self, key, value, expire=0, noreply=True):
+        with self.client_pool.get_and_release(destroy_on_fail=True) as client:
+            return client.replace(key, value, expire=expire, noreply=noreply)
+
+    def append(self, key, value, expire=0, noreply=True):
+        with self.client_pool.get_and_release(destroy_on_fail=True) as client:
+            return client.append(key, value, expire=expire, noreply=noreply)
+
+    def prepend(self, key, value, expire=0, noreply=True):
+        with self.client_pool.get_and_release(destroy_on_fail=True) as client:
+            return client.prepend(key, value, expire=expire, noreply=noreply)
+
+    def cas(self, key, value, cas, expire=0, noreply=False):
+        with self.client_pool.get_and_release(destroy_on_fail=True) as client:
+            return client.cas(key, value, cas,
+                              expire=expire, noreply=noreply)
+
+    def get(self, key):
+        with self.client_pool.get_and_release(destroy_on_fail=True) as client:
+            try:
+                return client.get(key)
+            except Exception:
+                if self.ignore_exc:
+                    return None
+                else:
+                    raise
+
+    def get_many(self, keys):
+        with self.client_pool.get_and_release(destroy_on_fail=True) as client:
+            try:
+                return client.get_many(keys)
+            except Exception:
+                if self.ignore_exc:
+                    return {}
+                else:
+                    raise
+
+    def gets(self, key):
+        with self.client_pool.get_and_release(destroy_on_fail=True) as client:
+            try:
+                return client.gets(key)
+            except Exception:
+                if self.ignore_exc:
+                    return (None, None)
+                else:
+                    raise
+
+    def gets_many(self, keys):
+        with self.client_pool.get_and_release(destroy_on_fail=True) as client:
+            try:
+                return client.gets_many(keys)
+            except Exception:
+                if self.ignore_exc:
+                    return {}
+                else:
+                    raise
+
+    def delete(self, key, noreply=True):
+        with self.client_pool.get_and_release(destroy_on_fail=True) as client:
+            return client.delete(key, noreply=noreply)
+
+    def delete_many(self, keys, noreply=True):
+        with self.client_pool.get_and_release(destroy_on_fail=True) as client:
+            return client.delete_many(keys, noreply=noreply)
+
+    def add(self, key, value, expire=0, noreply=True):
+        with self.client_pool.get_and_release(destroy_on_fail=True) as client:
+            return client.add(key, value, expire=expire, noreply=noreply)
+
+    def incr(self, key, value, noreply=False):
+        with self.client_pool.get_and_release(destroy_on_fail=True) as client:
+            return client.incr(key, value, noreply=noreply)
+
+    def decr(self, key, value, noreply=False):
+        with self.client_pool.get_and_release(destroy_on_fail=True) as client:
+            return client.decr(key, value, noreply=noreply)
+
+    def touch(self, key, expire=0, noreply=True):
+        with self.client_pool.get_and_release(destroy_on_fail=True) as client:
+            return client.touch(key, expire=expire, noreply=noreply)
+
+    def stats(self, *args):
+        with self.client_pool.get_and_release(destroy_on_fail=True) as client:
+            try:
+                return client.stats(*args)
+            except Exception:
+                if self.ignore_exc:
+                    return {}
+                else:
+                    raise
+
+    def flush_all(self, delay=0, noreply=True):
+        with self.client_pool.get_and_release(destroy_on_fail=True) as client:
+            return client.flush_all(delay=delay, noreply=noreply)
+
+    def quit(self):
+        with self.client_pool.get_and_release(destroy_on_fail=True) as client:
+            try:
+                client.quit()
+            finally:
+                self.client_pool.destroy(client)
+
+    def __setitem__(self, key, value):
+        self.set(key, value, noreply=True)
+
+    def __getitem__(self, key):
+        value = self.get(key)
+        if value is None:
+            raise KeyError
+        return value
+
+    def __delitem__(self, key):
+        self.delete(key, noreply=True)
+
+
 def _readline(sock, buf):
     """Read line of text from the socket.
 
diff --git a/pymemcache/pool.py b/pymemcache/pool.py
new file mode 100644
index 0000000..c341ceb
--- /dev/null
+++ b/pymemcache/pool.py
@@ -0,0 +1,110 @@
+# Copyright 2015 Yahoo.com
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import collections
+import contextlib
+import sys
+import threading
+
+import six
+
+
+class ObjectPool(object):
+    """A pool of objects that release/creates/destroys as needed."""
+
+    def __init__(self, obj_creator,
+                 after_remove=None, max_size=None):
+        self._used_objs = collections.deque()
+        self._free_objs = collections.deque()
+        self._obj_creator = obj_creator
+        self._lock = threading.Lock()
+        self._after_remove = after_remove
+        max_size = max_size or 2 ** 31
+        if not isinstance(max_size, six.integer_types) or max_size < 0:
+            raise ValueError('"max_size" must be a positive integer')
+        self.max_size = max_size
+
+    @property
+    def used(self):
+        return tuple(self._used_objs)
+
+    @property
+    def free(self):
+        return tuple(self._free_objs)
+
+    @contextlib.contextmanager
+    def get_and_release(self, destroy_on_fail=False):
+        obj = self.get()
+        try:
+            yield obj
+        except Exception:
+            exc_info = sys.exc_info()
+            if not destroy_on_fail:
+                self.release(obj)
+            else:
+                self.destroy(obj)
+            six.reraise(exc_info[0], exc_info[1], exc_info[2])
+        self.release(obj)
+
+    def get(self):
+        with self._lock:
+            if not self._free_objs:
+                curr_count = len(self._used_objs)
+                if curr_count >= self.max_size:
+                    raise RuntimeError("Too many objects,"
+                                       " %s >= %s" % (curr_count,
+                                                      self.max_size))
+                obj = self._obj_creator()
+                self._used_objs.append(obj)
+                return obj
+            else:
+                obj = self._free_objs.pop()
+                self._used_objs.append(obj)
+                return obj
+
+    def destroy(self, obj, silent=True):
+        was_dropped = False
+        with self._lock:
+            try:
+                self._used_objs.remove(obj)
+                was_dropped = True
+            except ValueError:
+                if not silent:
+                    raise
+        if was_dropped and self._after_remove is not None:
+            self._after_remove(obj)
+
+    def release(self, obj, silent=True):
+        with self._lock:
+            try:
+                self._used_objs.remove(obj)
+                self._free_objs.append(obj)
+            except ValueError:
+                if not silent:
+                    raise
+
+    def clear(self):
+        if self._after_remove is not None:
+            needs_destroy = []
+            with self._lock:
+                needs_destroy.extend(self._used_objs)
+                needs_destroy.extend(self._free_objs)
+                self._free_objs.clear()
+                self._used_objs.clear()
+            for obj in needs_destroy:
+                self._after_remove(obj)
+        else:
+            with self._lock:
+                self._free_objs.clear()
+                self._used_objs.clear()
diff --git a/pymemcache/test/test_client.py b/pymemcache/test/test_client.py
index 5aac8b8..c2837a8 100644
--- a/pymemcache/test/test_client.py
+++ b/pymemcache/test/test_client.py
@@ -18,9 +18,11 @@ import socket
 import unittest
 
 from nose import tools
+from pymemcache.client import PooledClient
 from pymemcache.client import Client, MemcacheUnknownCommandError
 from pymemcache.client import MemcacheClientError, MemcacheServerError
 from pymemcache.client import MemcacheUnknownError, MemcacheIllegalInputError
+from pymemcache import pool
 from pymemcache.test.utils import MockMemcacheClient
 
 
@@ -64,15 +66,18 @@ class MockSocketModule(object):
 
 
 class ClientTestMixin(object):
+    def make_client(self, mock_socket_values, serializer=None):
+        client = Client(None, serializer=serializer)
+        client.sock = MockSocket(list(mock_socket_values))
+        return client
+
     def test_set_success(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'STORED\r\n'])
+        client = self.make_client([b'STORED\r\n'])
         result = client.set(b'key', b'value', noreply=False)
         tools.assert_equal(result, True)
 
     def test_set_unicode_key(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b''])
+        client = self.make_client([b''])
 
         def _set():
             client.set(u'\u0FFF', b'value', noreply=False)
@@ -80,8 +85,7 @@ class ClientTestMixin(object):
         tools.assert_raises(MemcacheIllegalInputError, _set)
 
     def test_set_unicode_value(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b''])
+        client = self.make_client([b''])
 
         def _set():
             client.set(b'key', u'\u0FFF', noreply=False)
@@ -89,80 +93,69 @@ class ClientTestMixin(object):
         tools.assert_raises(MemcacheIllegalInputError, _set)
 
     def test_set_noreply(self):
-        client = self.Client(None)
-        client.sock = MockSocket([])
+        client = self.make_client([])
         result = client.set(b'key', b'value', noreply=True)
         tools.assert_equal(result, True)
 
     def test_set_many_success(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'STORED\r\n'])
+        client = self.make_client([b'STORED\r\n'])
         result = client.set_many({b'key' : b'value'}, noreply=False)
         tools.assert_equal(result, True)
 
     def test_add_stored(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'STORED\r', b'\n'])
+        client = self.make_client([b'STORED\r', b'\n'])
         result = client.add(b'key', b'value', noreply=False)
         tools.assert_equal(result, True)
 
     def test_add_not_stored(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'STORED\r', b'\n'])
+        client = self.make_client([b'STORED\r', b'\n',
+                                   b'NOT_', b'STOR', b'ED', b'\r\n'])
         result = client.add(b'key', b'value', noreply=False)
-
-        client.sock = MockSocket([b'NOT_', b'STOR', b'ED', b'\r\n'])
         result = client.add(b'key', b'value', noreply=False)
         tools.assert_equal(result, False)
 
     def test_get_not_found(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'END\r\n'])
+        client = self.make_client([b'END\r\n'])
         result = client.get(b'key')
         tools.assert_equal(result, None)
 
     def test_get_found(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'STORED\r\n'])
+        client = self.make_client([
+            b'STORED\r\n',
+            b'VALUE key 0 5\r\nvalue\r\nEND\r\n',
+        ])
         result = client.set(b'key', b'value', noreply=False)
-
-        client.sock = MockSocket([b'VALUE key 0 5\r\nvalue\r\nEND\r\n'])
         result = client.get(b'key')
         tools.assert_equal(result, b'value')
 
     def test_get_many_none_found(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'END\r\n'])
+        client = self.make_client([b'END\r\n'])
         result = client.get_many([b'key1', b'key2'])
         tools.assert_equal(result, {})
 
     def test_get_many_some_found(self):
-        client = self.Client(None)
-
-        client.sock = MockSocket([b'STORED\r\n'])
+        client = self.make_client([
+            b'STORED\r\n',
+            b'VALUE key1 0 6\r\nvalue1\r\nEND\r\n',
+        ])
         result = client.set(b'key1', b'value1', noreply=False)
-
-        client.sock = MockSocket([b'VALUE key1 0 6\r\nvalue1\r\nEND\r\n'])
         result = client.get_many([b'key1', b'key2'])
         tools.assert_equal(result, {b'key1': b'value1'})
 
     def test_get_many_all_found(self):
-        client = self.Client(None)
-
-        client.sock = MockSocket([b'STORED\r\n'])
+        client = self.make_client([
+            b'STORED\r\n',
+            b'STORED\r\n',
+            b'VALUE key1 0 6\r\nvalue1\r\n',
+            b'VALUE key2 0 6\r\nvalue2\r\nEND\r\n',
+        ])
         result = client.set(b'key1', b'value1', noreply=False)
-
-        client.sock = MockSocket([b'STORED\r\n'])
         result = client.set(b'key2', b'value2', noreply=False)
-
-        client.sock = MockSocket([b'VALUE key1 0 6\r\nvalue1\r\n'
-                                  b'VALUE key2 0 6\r\nvalue2\r\nEND\r\n'])
         result = client.get_many([b'key1', b'key2'])
         tools.assert_equal(result, {b'key1': b'value1', b'key2': b'value2'})
 
     def test_get_unicode_key(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b''])
+        client = self.make_client([b''])
 
         def _get():
             client.get(u'\u0FFF')
@@ -170,65 +163,48 @@ class ClientTestMixin(object):
         tools.assert_raises(MemcacheIllegalInputError, _get)
 
     def test_delete_not_found(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'NOT_FOUND\r\n'])
+        client = self.make_client([b'NOT_FOUND\r\n'])
         result = client.delete(b'key', noreply=False)
         tools.assert_equal(result, False)
 
     def test_delete_found(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'STORED\r', b'\n'])
+        client = self.make_client([b'STORED\r', b'\n', b'DELETED\r\n'])
         result = client.add(b'key', b'value', noreply=False)
-
-        client.sock = MockSocket([b'DELETED\r\n'])
         result = client.delete(b'key', noreply=False)
         tools.assert_equal(result, True)
 
     def test_delete_noreply(self):
-        client = self.Client(None)
-        client.sock = MockSocket([])
+        client = self.make_client([])
         result = client.delete(b'key', noreply=True)
         tools.assert_equal(result, True)
 
     def test_incr_not_found(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'NOT_FOUND\r\n'])
+        client = self.make_client([b'NOT_FOUND\r\n'])
         result = client.incr(b'key', 1, noreply=False)
         tools.assert_equal(result, None)
 
     def test_incr_found(self):
-        client = self.Client(None)
-
-        client.sock = MockSocket([b'STORED\r\n'])
+        client = self.make_client([b'STORED\r\n', b'1\r\n'])
         client.set(b'key', 0, noreply=False)
-
-        client.sock = MockSocket([b'1\r\n'])
         result = client.incr(b'key', 1, noreply=False)
         tools.assert_equal(result, 1)
 
     def test_incr_noreply(self):
-        client = self.Client(None)
-
-        client.sock = MockSocket([b'STORED\r\n'])
+        client = self.make_client([b'STORED\r\n'])
         client.set(b'key', 0, noreply=False)
 
-        client.sock = MockSocket([])
+        client = self.make_client([])
         result = client.incr(b'key', 1, noreply=True)
         tools.assert_equal(result, None)
 
     def test_decr_not_found(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'NOT_FOUND\r\n'])
+        client = self.make_client([b'NOT_FOUND\r\n'])
         result = client.decr(b'key', 1, noreply=False)
         tools.assert_equal(result, None)
 
     def test_decr_found(self):
-        client = self.Client(None)
-
-        client.sock = MockSocket([b'STORED\r\n'])
+        client = self.make_client([b'STORED\r\n', b'1\r\n'])
         client.set(b'key', 2, noreply=False)
-
-        client.sock = MockSocket([b'1\r\n'])
         result = client.decr(b'key', 1, noreply=False)
         tools.assert_equal(result, 1)
 
@@ -238,119 +214,107 @@ class TestClient(ClientTestMixin, unittest.TestCase):
     Client = Client
 
     def test_append_stored(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'STORED\r\n'])
+        client = self.make_client([b'STORED\r\n'])
         result = client.append(b'key', b'value', noreply=False)
         tools.assert_equal(result, True)
 
     def test_prepend_stored(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'STORED\r\n'])
+        client = self.make_client([b'STORED\r\n'])
         result = client.prepend(b'key', b'value', noreply=False)
         tools.assert_equal(result, True)
 
     def test_cas_stored(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'STORED\r\n'])
+        client = self.make_client([b'STORED\r\n'])
         result = client.cas(b'key', b'value', b'cas', noreply=False)
         tools.assert_equal(result, True)
 
     def test_cas_exists(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'EXISTS\r\n'])
+        client = self.make_client([b'EXISTS\r\n'])
         result = client.cas(b'key', b'value', b'cas', noreply=False)
         tools.assert_equal(result, False)
 
     def test_cas_not_found(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'NOT_FOUND\r\n'])
+        client = self.make_client([b'NOT_FOUND\r\n'])
         result = client.cas(b'key', b'value', b'cas', noreply=False)
         tools.assert_equal(result, None)
 
     def test_cr_nl_boundaries(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'VALUE key1 0 6\r',
-                                  b'\nvalue1\r\n'
-                                  b'VALUE key2 0 6\r\n',
-                                  b'value2\r\n'
-                                  b'END\r\n'])
+        client = self.make_client([b'VALUE key1 0 6\r',
+                                   b'\nvalue1\r\n'
+                                   b'VALUE key2 0 6\r\n',
+                                   b'value2\r\n'
+                                   b'END\r\n'])
         result = client.get_many([b'key1', b'key2'])
         tools.assert_equals(result, {b'key1': b'value1', b'key2': b'value2'})
 
-        client.sock = MockSocket([b'VALUE key1 0 6\r\n',
-                                  b'value1\r',
-                                  b'\nVALUE key2 0 6\r\n',
-                                  b'value2\r\n',
-                                  b'END\r\n'])
+        client = self.make_client([b'VALUE key1 0 6\r\n',
+                                   b'value1\r',
+                                   b'\nVALUE key2 0 6\r\n',
+                                   b'value2\r\n',
+                                   b'END\r\n'])
         result = client.get_many([b'key1', b'key2'])
         tools.assert_equals(result, {b'key1': b'value1', b'key2': b'value2'})
 
-        client.sock = MockSocket([b'VALUE key1 0 6\r\n',
-                                  b'value1\r\n',
-                                  b'VALUE key2 0 6\r',
-                                  b'\nvalue2\r\n',
-                                  b'END\r\n'])
+        client = self.make_client([b'VALUE key1 0 6\r\n',
+                                   b'value1\r\n',
+                                   b'VALUE key2 0 6\r',
+                                   b'\nvalue2\r\n',
+                                   b'END\r\n'])
         result = client.get_many([b'key1', b'key2'])
         tools.assert_equals(result, {b'key1': b'value1', b'key2': b'value2'})
 
 
-        client.sock = MockSocket([b'VALUE key1 0 6\r\n',
-                                  b'value1\r\n',
-                                  b'VALUE key2 0 6\r\n',
-                                  b'value2\r',
-                                  b'\nEND\r\n'])
+        client = self.make_client([b'VALUE key1 0 6\r\n',
+                                   b'value1\r\n',
+                                   b'VALUE key2 0 6\r\n',
+                                   b'value2\r',
+                                   b'\nEND\r\n'])
         result = client.get_many([b'key1', b'key2'])
         tools.assert_equals(result, {b'key1': b'value1', b'key2': b'value2'})
 
-        client.sock = MockSocket([b'VALUE key1 0 6\r\n',
-                                  b'value1\r\n',
-                                  b'VALUE key2 0 6\r\n',
-                                  b'value2\r\n',
-                                  b'END\r',
-                                  b'\n'])
+        client = self.make_client([b'VALUE key1 0 6\r\n',
+                                   b'value1\r\n',
+                                   b'VALUE key2 0 6\r\n',
+                                   b'value2\r\n',
+                                   b'END\r',
+                                   b'\n'])
         result = client.get_many([b'key1', b'key2'])
         tools.assert_equals(result, {b'key1': b'value1', b'key2': b'value2'})
 
-        client.sock = MockSocket([b'VALUE key1 0 6\r',
-                                  b'\nvalue1\r',
-                                  b'\nVALUE key2 0 6\r',
-                                  b'\nvalue2\r',
-                                  b'\nEND\r',
-                                  b'\n'])
+        client = self.make_client([b'VALUE key1 0 6\r',
+                                   b'\nvalue1\r',
+                                   b'\nVALUE key2 0 6\r',
+                                   b'\nvalue2\r',
+                                   b'\nEND\r',
+                                   b'\n'])
         result = client.get_many([b'key1', b'key2'])
         tools.assert_equals(result, {b'key1': b'value1', b'key2': b'value2'})
 
     def test_delete_exception(self):
-        client = self.Client(None)
-        client.sock = MockSocket([Exception('fail')])
+        client = self.make_client([Exception('fail')])
 
         def _delete():
             client.delete(b'key', noreply=False)
 
         tools.assert_raises(Exception, _delete)
         tools.assert_equal(client.sock, None)
-        tools.assert_equal(client.buf, b'')
 
     def test_flush_all(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'OK\r\n'])
+        client = self.make_client([b'OK\r\n'])
         result = client.flush_all(noreply=False)
         tools.assert_equal(result, True)
 
     def test_incr_exception(self):
-        client = self.Client(None)
-        client.sock = MockSocket([Exception('fail')])
+        client = self.make_client([Exception('fail')])
 
         def _incr():
             client.incr(b'key', 1)
 
         tools.assert_raises(Exception, _incr)
         tools.assert_equal(client.sock, None)
-        tools.assert_equal(client.buf, b'')
 
     def test_get_error(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'ERROR\r\n'])
+        client = self.make_client([b'ERROR\r\n'])
 
         def _get():
             client.get(b'key')
@@ -358,15 +322,13 @@ class TestClient(ClientTestMixin, unittest.TestCase):
         tools.assert_raises(MemcacheUnknownCommandError, _get)
 
     def test_get_recv_chunks(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'VALUE key', b' 0 5\r', b'\nvalue', b'\r\n',
-                                  b'END', b'\r', b'\n'])
+        client = self.make_client([b'VALUE key', b' 0 5\r', b'\nvalue',
+                                   b'\r\n', b'END', b'\r', b'\n'])
         result = client.get(b'key')
         tools.assert_equal(result, b'value')
 
     def test_get_unknown_error(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'foobarbaz\r\n'])
+        client = self.make_client([b'foobarbaz\r\n'])
 
         def _get():
             client.get(b'key')
@@ -374,58 +336,48 @@ class TestClient(ClientTestMixin, unittest.TestCase):
         tools.assert_raises(MemcacheUnknownError, _get)
 
     def test_gets_not_found(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'END\r\n'])
+        client = self.make_client([b'END\r\n'])
         result = client.gets(b'key')
         tools.assert_equal(result, (None, None))
 
     def test_gets_found(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'VALUE key 0 5 10\r\nvalue\r\nEND\r\n'])
+        client = self.make_client([b'VALUE key 0 5 10\r\nvalue\r\nEND\r\n'])
         result = client.gets(b'key')
         tools.assert_equal(result, (b'value', b'10'))
 
     def test_gets_many_none_found(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'END\r\n'])
+        client = self.make_client([b'END\r\n'])
         result = client.gets_many([b'key1', b'key2'])
         tools.assert_equal(result, {})
 
     def test_gets_many_some_found(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'VALUE key1 0 6 11\r\nvalue1\r\nEND\r\n'])
+        client = self.make_client([b'VALUE key1 0 6 11\r\nvalue1\r\nEND\r\n'])
         result = client.gets_many([b'key1', b'key2'])
         tools.assert_equal(result, {b'key1': (b'value1', b'11')})
 
     def test_touch_not_found(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'NOT_FOUND\r\n'])
+        client = self.make_client([b'NOT_FOUND\r\n'])
         result = client.touch(b'key', noreply=False)
         tools.assert_equal(result, False)
 
     def test_touch_found(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'TOUCHED\r\n'])
+        client = self.make_client([b'TOUCHED\r\n'])
         result = client.touch(b'key', noreply=False)
         tools.assert_equal(result, True)
 
     def test_quit(self):
-        client = self.Client(None)
-        client.sock = MockSocket([])
+        client = self.make_client([])
         result = client.quit()
         tools.assert_equal(result, None)
         tools.assert_equal(client.sock, None)
-        tools.assert_equal(client.buf, b'')
 
     def test_replace_stored(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'STORED\r\n'])
+        client = self.make_client([b'STORED\r\n'])
         result = client.replace(b'key', b'value', noreply=False)
         tools.assert_equal(result, True)
 
     def test_replace_not_stored(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'NOT_STORED\r\n'])
+        client = self.make_client([b'NOT_STORED\r\n'])
         result = client.replace(b'key', b'value', noreply=False)
         tools.assert_equal(result, False)
 
@@ -433,24 +385,21 @@ class TestClient(ClientTestMixin, unittest.TestCase):
         def _ser(key, value):
             return json.dumps(value), 0
 
-        client = self.Client(None, serializer=_ser)
-        client.sock = MockSocket([b'STORED\r\n'])
+        client = self.make_client([b'STORED\r\n'], serializer=_ser)
         client.set('key', {'c': 'd'})
         tools.assert_equal(client.sock.send_bufs, [
             b'set key 0 0 10 noreply\r\n{"c": "d"}\r\n'
         ])
 
     def test_set_socket_handling(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'STORED\r\n'])
+        client = self.make_client([b'STORED\r\n'])
         result = client.set(b'key', b'value', noreply=False)
         tools.assert_equal(result, True)
         tools.assert_equal(client.sock.closed, False)
         tools.assert_equal(len(client.sock.send_bufs), 1)
 
     def test_set_error(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'ERROR\r\n'])
+        client = self.make_client([b'ERROR\r\n'])
 
         def _set():
             client.set(b'key', b'value', noreply=False)
@@ -458,19 +407,16 @@ class TestClient(ClientTestMixin, unittest.TestCase):
         tools.assert_raises(MemcacheUnknownCommandError, _set)
 
     def test_set_exception(self):
-        client = self.Client(None)
-        client.sock = MockSocket([Exception('fail')])
+        client = self.make_client([Exception('fail')])
 
         def _set():
             client.set(b'key', b'value', noreply=False)
 
         tools.assert_raises(Exception, _set)
         tools.assert_equal(client.sock, None)
-        tools.assert_equal(client.buf, b'')
 
     def test_set_client_error(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'CLIENT_ERROR some message\r\n'])
+        client = self.make_client([b'CLIENT_ERROR some message\r\n'])
 
         def _set():
             client.set('key', 'value', noreply=False)
@@ -478,8 +424,7 @@ class TestClient(ClientTestMixin, unittest.TestCase):
         tools.assert_raises(MemcacheClientError, _set)
 
     def test_set_server_error(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'SERVER_ERROR some message\r\n'])
+        client = self.make_client([b'SERVER_ERROR some message\r\n'])
 
         def _set():
             client.set(b'key', b'value', noreply=False)
@@ -487,8 +432,7 @@ class TestClient(ClientTestMixin, unittest.TestCase):
         tools.assert_raises(MemcacheServerError, _set)
 
     def test_set_unknown_error(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'foobarbaz\r\n'])
+        client = self.make_client([b'foobarbaz\r\n'])
 
         def _set():
             client.set(b'key', b'value', noreply=False)
@@ -496,16 +440,14 @@ class TestClient(ClientTestMixin, unittest.TestCase):
         tools.assert_raises(MemcacheUnknownError, _set)
 
     def test_set_many_socket_handling(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'STORED\r\n'])
+        client = self.make_client([b'STORED\r\n'])
         result = client.set_many({b'key': b'value'}, noreply=False)
         tools.assert_equal(result, True)
         tools.assert_equal(client.sock.closed, False)
         tools.assert_equal(len(client.sock.send_bufs), 1)
 
     def test_set_many_exception(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'STORED\r\n', Exception('fail')])
+        client = self.make_client([b'STORED\r\n', Exception('fail')])
 
         def _set():
             client.set_many({b'key': b'value', b'other': b'value'},
@@ -513,11 +455,9 @@ class TestClient(ClientTestMixin, unittest.TestCase):
 
         tools.assert_raises(Exception, _set)
         tools.assert_equal(client.sock, None)
-        tools.assert_equal(client.buf, b'')
 
     def test_stats(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'STAT fake_stats 1\r\n', b'END\r\n'])
+        client = self.make_client([b'STAT fake_stats 1\r\n', b'END\r\n'])
         result = client.stats()
         tools.assert_equal(client.sock.send_bufs, [
             b'stats \r\n'
@@ -525,8 +465,7 @@ class TestClient(ClientTestMixin, unittest.TestCase):
         tools.assert_equal(result, {b'fake_stats': 1})
 
     def test_stats_with_args(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'STAT fake_stats 1\r\n', b'END\r\n'])
+        client = self.make_client([b'STAT fake_stats 1\r\n', b'END\r\n'])
         result = client.stats('some_arg')
         tools.assert_equal(client.sock.send_bufs, [
             b'stats some_arg\r\n'
@@ -534,8 +473,7 @@ class TestClient(ClientTestMixin, unittest.TestCase):
         tools.assert_equal(result, {b'fake_stats': 1})
 
     def test_stats_conversions(self):
-        client = self.Client(None)
-        client.sock = MockSocket([
+        client = self.make_client([
             # Most stats are converted to int
             b'STAT cmd_get 2519\r\n',
             b'STAT cmd_set 3099\r\n',
@@ -567,42 +505,16 @@ class TestClient(ClientTestMixin, unittest.TestCase):
         }
         tools.assert_equal(result, expected)
 
-    def test_socket_connect(self):
-        server = ("example.com", 11211)
-
-        client = Client(server, socket_module=MockSocketModule())
-        client._connect()
-        tools.assert_equal(client.sock.connections, [server])
-
-        timeout = 2
-        connect_timeout = 3
-        client = Client(server, connect_timeout=connect_timeout, timeout=timeout,
-                        socket_module=MockSocketModule())
-        client._connect()
-        tools.assert_equal(client.sock.timeouts, [connect_timeout, timeout])
-
-        client = Client(server, socket_module=MockSocketModule())
-        client._connect()
-        tools.assert_equal(client.sock.socket_options, [])
-
-        client = Client(server, socket_module=MockSocketModule(), no_delay=True)
-        client._connect()
-        tools.assert_equal(client.sock.socket_options, [(socket.IPPROTO_TCP,
-                                                        socket.TCP_NODELAY, 1)])
-
     def test_python_dict_set_is_supported(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'STORED\r\n'])
+        client = self.make_client([b'STORED\r\n'])
         client[b'key'] = b'value'
 
     def test_python_dict_get_is_supported(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'VALUE key 0 5\r\nvalue\r\nEND\r\n'])
+        client = self.make_client([b'VALUE key 0 5\r\nvalue\r\nEND\r\n'])
         tools.assert_equal(client[b'key'], b'value')
 
     def test_python_dict_get_not_found_is_supported(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'END\r\n'])
+        client = self.make_client([b'END\r\n'])
 
         def _get():
             _ = client[b'key']
@@ -610,68 +522,108 @@ class TestClient(ClientTestMixin, unittest.TestCase):
         tools.assert_raises(KeyError, _get)
 
     def test_python_dict_del_is_supported(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'DELETED\r\n'])
+        client = self.make_client([b'DELETED\r\n'])
         del client[b'key']
 
     def test_too_long_key(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'END\r\n'])
+        client = self.make_client([b'END\r\n'])
         tools.assert_raises(MemcacheClientError, client.get, b'x' * 251)
 
     def test_key_contains_spae(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'END\r\n'])
+        client = self.make_client([b'END\r\n'])
         tools.assert_raises(MemcacheClientError, client.get, b'abc xyz')
 
     def test_key_contains_nonascii(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'END\r\n'])
+        client = self.make_client([b'END\r\n'])
         tools.assert_raises(MemcacheClientError, client.get, u'\u3053\u3093\u306b\u3061\u306f')
 
 
+class TestClientSocketConnect(unittest.TestCase):
+    def test_socket_connect(self):
+        server = ("example.com", 11211)
+
+        client = Client(server, socket_module=MockSocketModule())
+        client._connect()
+        tools.assert_equal(client.sock.connections, [server])
+
+        timeout = 2
+        connect_timeout = 3
+        client = Client(server, connect_timeout=connect_timeout, timeout=timeout,
+                        socket_module=MockSocketModule())
+        client._connect()
+        tools.assert_equal(client.sock.timeouts, [connect_timeout, timeout])
+
+        client = Client(server, socket_module=MockSocketModule())
+        client._connect()
+        tools.assert_equal(client.sock.socket_options, [])
+
+        client = Client(server, socket_module=MockSocketModule(), no_delay=True)
+        client._connect()
+        tools.assert_equal(client.sock.socket_options, [(socket.IPPROTO_TCP,
+                                                        socket.TCP_NODELAY, 1)])
+
+
+class TestPooledClient(ClientTestMixin, unittest.TestCase):
+    def make_client(self, mock_socket_values, serializer=None):
+        mock_client = Client(None, serializer=serializer)
+        mock_client.sock = MockSocket(list(mock_socket_values))
+        client = PooledClient(None, serializer=serializer)
+        client.client_pool = pool.ObjectPool(lambda: mock_client)
+        return client
+
+
 class TestMockClient(ClientTestMixin, unittest.TestCase):
-    Client = MockMemcacheClient
+    def make_client(self, mock_socket_values, serializer=None):
+        client = MockMemcacheClient(None, serializer=serializer)
+        client.sock = MockSocket(list(mock_socket_values))
+        return client
 
 
 class TestPrefixedClient(ClientTestMixin, unittest.TestCase):
-    def Client(self, *args, **kwargs):
-        return Client(*args, key_prefix=b'xyz:', **kwargs)
+    def make_client(self, mock_socket_values, serializer=None):
+        client = Client(None, serializer=serializer, key_prefix=b'xyz:')
+        client.sock = MockSocket(list(mock_socket_values))
+        return client
 
     def test_get_found(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'STORED\r\n'])
+        client = self.make_client([
+            b'STORED\r\n',
+            b'VALUE xyz:key 0 5\r\nvalue\r\nEND\r\n',
+        ])
         result = client.set(b'key', b'value', noreply=False)
-
-        client.sock = MockSocket([b'VALUE xyz:key 0 5\r\nvalue\r\nEND\r\n'])
         result = client.get(b'key')
         tools.assert_equal(result, b'value')
 
     def test_get_many_some_found(self):
-        client = self.Client(None)
-
-        client.sock = MockSocket([b'STORED\r\n'])
+        client = self.make_client([
+            b'STORED\r\n',
+            b'VALUE xyz:key1 0 6\r\nvalue1\r\nEND\r\n',
+        ])
         result = client.set(b'key1', b'value1', noreply=False)
-
-        client.sock = MockSocket([b'VALUE xyz:key1 0 6\r\nvalue1\r\nEND\r\n'])
         result = client.get_many([b'key1', b'key2'])
         tools.assert_equal(result, {b'key1': b'value1'})
 
     def test_get_many_all_found(self):
-        client = self.Client(None)
-
-        client.sock = MockSocket([b'STORED\r\n'])
+        client = self.make_client([
+            b'STORED\r\n',
+            b'STORED\r\n',
+            b'VALUE xyz:key1 0 6\r\nvalue1\r\n',
+            b'VALUE xyz:key2 0 6\r\nvalue2\r\nEND\r\n',
+        ])
         result = client.set(b'key1', b'value1', noreply=False)
-
-        client.sock = MockSocket([b'STORED\r\n'])
         result = client.set(b'key2', b'value2', noreply=False)
-
-        client.sock = MockSocket([b'VALUE xyz:key1 0 6\r\nvalue1\r\n'
-                                  b'VALUE xyz:key2 0 6\r\nvalue2\r\nEND\r\n'])
         result = client.get_many([b'key1', b'key2'])
         tools.assert_equal(result, {b'key1': b'value1', b'key2': b'value2'})
 
     def test_python_dict_get_is_supported(self):
-        client = self.Client(None)
-        client.sock = MockSocket([b'VALUE xyz:key 0 5\r\nvalue\r\nEND\r\n'])
+        client = self.make_client([b'VALUE xyz:key 0 5\r\nvalue\r\nEND\r\n'])
         tools.assert_equal(client[b'key'], b'value')
+
+
+class TestPrefixedPooledClient(TestPrefixedClient):
+    def make_client(self, mock_socket_values, serializer=None):
+        mock_client = Client(None, serializer=serializer, key_prefix=b'xyz:')
+        mock_client.sock = MockSocket(list(mock_socket_values))
+        client = PooledClient(None, serializer=serializer, key_prefix=b'xyz:')
+        client.client_pool = pool.ObjectPool(lambda: mock_client)
+        return client