python-memcached的线程安全问题

答案是肯定的，前提你在使用Python 2.4+和python-memcached 1.36+
为什么我们需要线程安全的memcached client，因为我们的实际应用一般是多线程的模型，例如cherrypy、twisted，如果python-memcached不是线程安全的话， 引起的问题不仅仅是并发修改共享变量这么简单，是外部socket链接的数据流的混乱
python-memcached怎么实现线程安全的呢？查看源代码看到

try:
    # Only exists in Python 2.4+
    from threading import local
except ImportError:
    # TODO:  add the pure-python local implementation
    class local(object):
        pass

 class Client(local): 很取巧的让Client类继承threading.local，也就是Client里面的每一个属性都是跟当前线程绑定的。实现虽然不太优雅，但是很实在
但是别以为这样就可以随便在线程里面用python-memcached了，因为这种thread local的做法，你的应用必须要使用thread pool的模式，而不能不停创建销毁thread，因为每一个新线程的创建，对于就会使用一个全新的Client，也就是一个全新的socket链接，如 果不停打开创建销毁thread的话，就会导致不停的创建销毁socket链接，导致性能大量下降。幸好，无论是cherrypy还是twisted，都 是使用了thread pool的模式。

 

但是不幸的是gevent不是thread pool的模式，这导致不停的创建销毁socket链接。

gevent

# ThreadID: 47316030715200
=============================
File: "/data1/duitang/dist/app/test/7199/duitang/common/templatetags/myTags.py", line 525, in statichtml
return StaticHTML_Tag(static_name)
File: "/data1/duitang/dist/app/test/7199/duitang/common/templatetags/myTags.py", line 507, in __init__
sh = StaticHTML.objects.get(name=name)
File: "/data1/duitang/dist/app/test/7199/duitang/statichtml/manager.py", line 18, in get
model = key and cache.get(key)
File: "/data1/duitang/dist/app/test/7199/duitang/perf/memcached.py", line 25, in get
rp = super(MemcachedCache, self).get(key, deault, version)
File: "/duitang/dist/sys/python/lib/python2.7/site-packages/django/core/cache/backends/memcached.py", line 58, in get
val = self._cache.get(key)
File: "build/bdist.linux-x86_64/egg/gevent/local.py", line 180, in __getattribute__
_init_locals(self)
File: "build/bdist.linux-x86_64/egg/gevent/local.py", line 167, in _init_locals
cls.__init__(self, *args, **kw)
File: "/data1/duitang/dist/app/test/7199/duitang/memcache.py", line 168, in __init__
print cf1.stacktraces()
File: "/data1/duitang/dist/app/test/7199/duitang/cf1.py", line 8, in stacktraces
for filename, lineno, name, line in traceback.extract_stack(stack):

========================================================
# ThreadID: 47316030715200
File: "build/bdist.linux-x86_64/egg/gevent/greenlet.py", line 390, in run
  result = self._run(*self.args, **self.kwargs)
File: "/duitang/dist/sys/python/lib/python2.7/site-packages/gunicorn-0.14.6-py2.7.egg/gunicorn/workers/async.py", line 44, in handle
  self.handle_request(req, client, addr)
File: "/duitang/dist/sys/python/lib/python2.7/site-packages/gunicorn-0.14.6-py2.7.egg/gunicorn/workers/ggevent.py", line 88, in handle_request
  super(GeventWorker, self).handle_request(*args)
File: "/duitang/dist/sys/python/lib/python2.7/site-packages/gunicorn-0.14.6-py2.7.egg/gunicorn/workers/async.py", line 78, in handle_request
  respiter = self.wsgi(environ, resp.start_response)
File: "/duitang/dist/sys/python/lib/python2.7/site-packages/django/core/handlers/wsgi.py", line 273, in __call__
  response = self.get_response(request)
File: "/duitang/dist/sys/python/lib/python2.7/site-packages/django/core/handlers/base.py", line 111, in get_response
  response = callback(request, *callback_args, **callback_kwargs)
File: "/duitang/dist/sys/python/lib/python2.7/site-packages/django/utils/decorators.py", line 93, in _wrapped_view
  response = view_func(request, *args, **kwargs)
File: "/data1/duitang/dist/app/test/7199/duitang/people/views_people.py", line 45, in peopleIndex
  people = AuthUser.objects.get(id=UserProfile.get_real_id(user_id))
File: "/data1/duitang/dist/app/test/7199/duitang/common/user.py", line 19, in get
  model = key and cache.get(key)
File: "/data1/duitang/dist/app/test/7199/duitang/perf/memcached.py", line 25, in get
  rp = super(MemcachedCache, self).get(key, deault, version)
File: "/duitang/dist/sys/python/lib/python2.7/site-packages/django/core/cache/backends/memcached.py", line 58, in get
  val = self._cache.get(key)
File: "build/bdist.linux-x86_64/egg/gevent/local.py", line 180, in __getattribute__
  _init_locals(self)
File: "build/bdist.linux-x86_64/egg/gevent/local.py", line 167, in _init_locals
  cls.__init__(self, *args, **kw)
File: "/data1/duitang/dist/app/test/7199/duitang/memcache.py", line 168, in __init__
  print cf1.stacktraces()
File: "/data1/duitang/dist/app/test/7199/duitang/cf1.py", line 8, in stacktraces
  for filename, lineno, name, line in traceback.extract_stack(stack):
   

可以发现threading local最后委托给gevent/local.py实现，导致重复初始化Client。

 

而uwsgi只有初始化时才调用，后续刷新页面不再触发Client.init()

 

google了一下，老外的解决方案： https://groups.google.com/forum/?fromgroups=#!topic/gevent/ULKUPvbaQ7I

 

 

 

 

python memcache客户端比较

对于memcached，redis,hessian,mongo等各种客户端都要考虑一个问题，如何高效安全地和server进行通信？有几种策略：

1. 每次调用都建立和关闭连接
优点：实现简单，不存在线程安全问题。django的BaseMemcachedCache实现就是每次执行完都会close掉连接。
缺点：出现大量TIME_WAIT,花费在建立网络连接上的开销比较大，在有些情况下会导致性能瓶颈。

2.单一socket长线程
优点：不会出现大量TIME_WAIT,不会频繁开关socket,减轻服务器的压力。
缺点：远程调用一般是请求/问答式，为了保证线程安全必须加同步锁，如果远程服务器响应慢的话会导致大量线程竞争同一个socket资源，socket成为瓶颈。

3.单一socket长线程+连接池
为了解决方案1和方案2的问题，我们采用方案3：保持长连接不变，但采用连接池来保存长连接，这样系统内有一批socket供程序调用，避免大量线程竞争同一个socket。

那么python memcached client采用那种形式呢？

首先python memcached client有好几个,在memcached的官方网站提供了client列表(http://code.google.com/p/memcached/wiki/Clients)：
a.libmemcached: 最受欢迎的memcached的C语言版本的客户端的，高性能，线程安全。很多语言都有对这个版本的wrapper。
b.pylibmc:是对libmemcached的wrapper。 http://sendapatch.se/projects/pylibmc/
c.python-memcached: 是100%纯python的版本，也是我们现在正在使用的版本。     http://www.tummy.com/Community/software/python-memcached/
d.Python libmemcached: 豆瓣提供的，对libmemcached的wrapper版本。http://code.google.com/p/python-libmemcached/
e.django cahce: django对memcached的支持，其实django只是提供了一个统一的cache接口,并没有具体实现，相关实现类在django.core.cache.backends.memcached.BaseMemcachedCache ，具体参考：https://docs.djangoproject.com/en/dev/topics/cache/ 不过django的问题很多。

python-memcached
python-memcached不会在每次get/set操作完成之后主动关闭连接，他是一种长连接，但他如果保证线程安全呢？一般我们是这样使用它：
import memcache
mc = memcache.Client(['127.0.0.1:11211'], debug=0)
答案在于他很取巧的让Client类继承threading.local，也就是Client里面的每一个属性都是跟当前线程绑定。这样每个线程都只会看到本地的Client，变相的实现了连接池。
这个池的大小取决于系统有多少个线程。为了验证这个说法，我们写一个程序来测试。
[code]
import memcache
import threading
mc = memcache.Client(['127.0.0.1:11211'], debug=1)

class TestThread(threading.Thread): 
    def __init__(self,redis_cache): 
        threading.Thread.__init__(self) 
        self.redis_cache = redis_cache 
 
    def run(self): 
        while True: 
            obj = self.redis_cache.get("uid:1002")
            if not obj:
                self.redis_cache.set("uid:1002","test",1)
            print obj

for i in xrange(8): 
    t = TestThread(mc) 
    t.start()   
[/code]

执行这段程序，然后通过netstat -an查看，刚好有一个8个ESTABLISHED，每个进程一个socket。
python的locale和java的ThreadLocl是一个意思，把值设置到线程中，每个线程只能看到自己线程保存的值，这里面的实现是这样的，由于Client继承于local，当访问Cient的任何一个方法或者属性都会进入到__getattribute__()，其内部实现是获取到当前线程并绑定 相关代码如下：

def _patch(self):
    key = object.__getattribute__(self, '_local__key')
    d = current_thread().__dict__.get(key)
    if d is None:
        d = {}
        current_thread().__dict__[key] = d
        object.__setattr__(self, '__dict__', d)

        # we have a new instance dict, so call out __init__ if we have
        # one
        cls = type(self)
        if cls.__init__ is not object.__init__:
            args, kw = object.__getattribute__(self, '_local__args')
            cls.__init__(self, *args, **kw)
    else:
        object.__setattr__(self, '__dict__', d)

 

cls是当前对象的class，object是基类class，我打印了清单：

new :  <class 'duitang.memcache.Client'> | <type 'object'>
--------------------
cached: <class 'gevent.local.local'> | <slot wrapper '__init__' of 'object' objects> | <slot wrapper '__init__' of 'object' objects>

yunpeng@yunpeng-duitang:~/test2$ netstat -an | grep 11211
tcp        0      0 0.0.0.0:11211           0.0.0.0:*               LISTEN    
tcp       31      0 127.0.0.1:46835         127.0.0.1:11211         ESTABLISHED
tcp        0      0 127.0.0.1:46832         127.0.0.1:11211         ESTABLISHED
tcp        0      0 127.0.0.1:46831         127.0.0.1:11211         ESTABLISHED
tcp        0      0 127.0.0.1:46829         127.0.0.1:11211         ESTABLISHED
tcp        0     14 127.0.0.1:46833         127.0.0.1:11211         ESTABLISHED
tcp       31      0 127.0.0.1:46828         127.0.0.1:11211         ESTABLISHED
tcp        0      0 127.0.0.1:46834         127.0.0.1:11211         ESTABLISHED
tcp       31      0 127.0.0.1:46830         127.0.0.1:11211         ESTABLISHED

当python进程退出socket会被自动关闭。
yunpeng@yunpeng-duitang:/duitang/dist/app/trunk/duitang$ netstat -an | grep 11211
tcp        0      0 0.0.0.0:11211           0.0.0.0:*               LISTEN    
tcp        0      0 127.0.0.1:44900         127.0.0.1:11211         TIME_WAIT 


但是很明显，采用threading locale这种方式来保证线程安全存在一些缺陷：

1.要求web server采用thread pool的方式，如果thread每次执行完之后就结束了，这会导致不停的创建销毁socket链接。
2.要求使用python thread locale的语义，但不幸的是python的thread语义很容易被改变，gevent就可以直接把python的一个thread转换成greenlet。


gevent的monkey提供的patch方法

patch_all() 调用所有的monkey patch
   
patch_os() os.fork()替换成gevent.fork


patch_select(aggressive=False) `select.select`替换成`gevent.select.select`

patch_socket(dns=True, aggressive=True)  标准的socket object 替换成 gevent's cooperative sockets.

patch_thread(threading=True, _threading_local=True) thread` module 替换成 gevent's thread

patch_time()  把标准的`time.sleep` 替换成`gevent.sleep`.


gunicorn如何使用gevent?
代码：
/duitang/dist/sys/python/lib/python2.7/site-packages/gunicorn-0.14.6-py2.7.egg/gunicorn/workers/ggevent.py

# -*- coding: utf-8 -
#
# This file is part of gunicorn released under the MIT license.
# See the NOTICE for more information.

from __future__ import with_statement

import os
import sys
from datetime import datetime

# workaround on osx, disable kqueue
if sys.platform == "darwin":
    os.environ['EVENT_NOKQUEUE'] = "1"

try:
    import gevent
except ImportError:
    raise RuntimeError("You need gevent installed to use this worker.")
from gevent.pool import Pool
from gevent.server import StreamServer
from gevent import pywsgi

import gunicorn
from gunicorn.workers.async import AsyncWorker

VERSION = "gevent/%s gunicorn/%s" % (gevent.__version__, gunicorn.__version__)

BASE_WSGI_ENV = {
    'GATEWAY_INTERFACE': 'CGI/1.1',
    'SERVER_SOFTWARE': VERSION,
    'SCRIPT_NAME': '',
    'wsgi.version': (1, 0),
    'wsgi.multithread': False,
    'wsgi.multiprocess': False,
    'wsgi.run_once': False
}

class GeventWorker(AsyncWorker):

    server_class = None
    wsgi_handler = None

    @classmethod
    def setup(cls):
        from gevent import monkey
        monkey.noisy = False
        monkey.patch_all()

    def timeout_ctx(self):
        return gevent.Timeout(self.cfg.keepalive, False)

    def run(self):
        self.socket.setblocking(1)

        pool = Pool(self.worker_connections)
        if self.server_class is not None:
            server = self.server_class(
                self.socket, application=self.wsgi, spawn=pool, log=self.log,
                handler_class=self.wsgi_handler)
        else:
            server = StreamServer(self.socket, handle=self.handle, spawn=pool)

        server.start()
        pid = os.getpid()
        try:
            while self.alive:
                self.notify()

                if  pid == os.getpid() and self.ppid != os.getppid():
                    self.log.info("Parent changed, shutting down: %s", self)
                    break

                gevent.sleep(1.0)

        except KeyboardInterrupt:
            pass

        try:
            # Try to stop connections until timeout
            self.notify()
            server.stop(timeout=self.cfg.graceful_timeout)
        except:
            pass

    def handle_request(self, *args):
        try:
            super(GeventWorker, self).handle_request(*args)
        except gevent.GreenletExit:
            pass

        if gevent.version_info[0] == 0:

            def init_process(self):
                #gevent 0.13 and older doesn't reinitialize dns for us after forking
                #here's the workaround
                import gevent.core
                gevent.core.dns_shutdown(fail_requests=1)
                gevent.core.dns_init()
                super(GeventWorker, self).init_process()


class GeventResponse(object):

    status = None
    headers = None
    response_length = None


    def __init__(self, status, headers, clength):
        self.status = status
        self.headers = headers
        self.response_length = clength

class PyWSGIHandler(pywsgi.WSGIHandler):

    def log_request(self):
        start = datetime.fromtimestamp(self.time_start)
        finish = datetime.fromtimestamp(self.time_finish)
        response_time = finish - start
        resp = GeventResponse(self.status, self.response_headers,
                self.response_length)
        req_headers = [h.split(":", 1) for h in self.headers.headers]
        self.server.log.access(resp, req_headers, self.environ, response_time)

    def get_environ(self):
        env = super(PyWSGIHandler, self).get_environ()
        env['gunicorn.sock'] = self.socket
        env['RAW_URI'] = self.path
        return env

class PyWSGIServer(pywsgi.WSGIServer):
    base_env = BASE_WSGI_ENV

class GeventPyWSGIWorker(GeventWorker):
    "The Gevent StreamServer based workers."
    server_class = PyWSGIServer
    wsgi_handler = PyWSGIHandler

 

测试：ab -n100 http://7199.t.duitang.com:7199/cache/

[admin@server2 duitang]$ netstat -an| grep 11211 |wc -l
306

总结：使用了gevent之后thread local有太多不可控.
gevent代码：/duitang/dist/sys/python/lib/python2.7/site-packages/gevent
gevent+django:http://www.slideshare.net/mahendram/scaling-django-with-gevent