python-multithreading - gRPC + Thread 本地问题

Question

我用 python 构建了一个 grpc 服务器，并尝试使用 werkzeug Local 和 LocalProxy 处理一些线程本地存储，类似于 flask 所做的。

我面临的问题是，当我从服务器拦截器将一些数据存储在本地，然后尝试从服务器检索它时，本地是空的。真正的问题是，出于某种原因，拦截器在与服务程序不同的 greenlet 中运行，因此不可能在请求中共享数据，因为 werkzeug.local.storage 最终以不同的 key 结束属于同一个请求的数据。

使用 python 线程库也会发生同样的情况，看起来拦截器是从主线程或与服务程序不同的线程运行的。有解决方法吗？我希望拦截器在同一个线程中运行，从而允许此类事情发生。

# Define a global somewhere
from werkzeug.local import Local
local = Local()

# from an interceptor save something
local.message = "test msg"

# from the service access it
local.service_var = "test"
print local.message  # this throw a AttributeError

# print the content of local
print local.__storage__  # we have 2 entries in the storage, 2 different greenlets, but we are in the same request.

Answer 1

拦截器确实运行在与处理线程不同的服务线程上。服务线程负责服务服务和拦截服务处理程序。拦截器返回服务方法处理程序后，服务线程会将其提交到位于_server.py#L525 的thread_pool。 :

# Take unary unary call as an example.
# The method_handler is the returned object from interceptor.
def _handle_unary_unary(rpc_event, state, method_handler, thread_pool):
    unary_request = _unary_request(rpc_event, state,
                                   method_handler.request_deserializer)
    return thread_pool.submit(_unary_response_in_pool, rpc_event, state,
                              method_handler.unary_unary, unary_request,
                              method_handler.request_deserializer,
                              method_handler.response_serializer)

至于解决方法，我只能想象在初始化期间将存储实例同时传递给拦截器和服务程序。之后，存储可以用作成员变量。

class StorageServerInterceptor(grpc.ServerInterceptor):

    def __init__(self, storage):
        self._storage = storage

    def intercept_service(self, continuation, handler_call_details):
        key = ...
        value = ...
        self._storage.set(key, value)
        ...
        return continuation(handler_call_details)

class Storage(...StorageServicer):

    def __init__(self, storage):
        self._storage = storage

    ...Servicer Handlers...