Saltar al contenido

¿Cómo implementar un servidor python async grpc?

Contamos con tu apoyo para compartir nuestros ensayos en referencia a las ciencias de la computación.

Solución:

Como señaló @Michael en un comentario, a partir de la versión 1.32, gRPC ahora admite asyncio en su API de Python. Si está usando una versión anterior, aún puede usar la API asyncio a través de la API experimental: from grpc.experimental import aio. También se ha agregado un ejemplo de asyncio hello world al repositorio de gRPC. El siguiente código es una copia del servidor de ejemplo:

import logging                                                                  
import asyncio                                                                  
from grpc import aio                                                            
                                                                                
import helloworld_pb2                                                           
import helloworld_pb2_grpc                                                      
                                                                                
                                                                                
class Greeter(helloworld_pb2_grpc.GreeterServicer):                             
                                                                                
    async def SayHello(self, request, context):                                 
        return helloworld_pb2.HelloReply(message='Hello, %s!' % request.name)    
                                                                                
                                                                                
async def serve():                                                              
    server = aio.server()                                                       
    helloworld_pb2_grpc.add_GreeterServicer_to_server(Greeter(), server)        
    listen_addr = '[::]:50051'                                                  
    server.add_insecure_port(listen_addr)                                       
    logging.info("Starting server on %s", listen_addr)                          
    await server.start()                                                        
    await server.wait_for_termination()                                         
                                                                                
                                                                                
if __name__ == '__main__':                                                      
    logging.basicConfig(level=logging.INFO)                                     
    asyncio.run(serve())

Vea mi otra respuesta sobre cómo implementar el cliente.

En mi opinión, es una buena implementación simple del servidor async grpc, igual que basado en http en aiohttp.

import asyncio
from concurrent import futures
import functools
import inspect
import threading

from grpc import _server

def _loop_mgr(loop: asyncio.AbstractEventLoop):

    asyncio.set_event_loop(loop)
    loop.run_forever()

    # If we reach here, the loop was stopped.
    # We should gather any remaining tasks and finish them.
    pending = asyncio.Task.all_tasks(loop=loop)
    if pending:
        loop.run_until_complete(asyncio.gather(*pending))


class AsyncioExecutor(futures.Executor):

    def __init__(self, *, loop=None):

        super().__init__()
        self._shutdown = False
        self._loop = loop or asyncio.get_event_loop()
        self._thread = threading.Thread(target=_loop_mgr, args=(self._loop,),
                                        daemon=True)
        self._thread.start()

    def submit(self, fn, *args, **kwargs):

        if self._shutdown:
            raise RuntimeError('Cannot schedule new futures after shutdown')

        if not self._loop.is_running():
            raise RuntimeError("Loop must be started before any function can "
                               "be submitted")

        if inspect.iscoroutinefunction(fn):
            coro = fn(*args, **kwargs)
            return asyncio.run_coroutine_threadsafe(coro, self._loop)

        else:
            func = functools.partial(fn, *args, **kwargs)
            return self._loop.run_in_executor(None, func)

    def shutdown(self, wait=True):
        self._loop.stop()
        self._shutdown = True
        if wait:
            self._thread.join()


# --------------------------------------------------------------------------- #


async def _call_behavior(rpc_event, state, behavior, argument, request_deserializer):
    context = _server._Context(rpc_event, state, request_deserializer)
    try:
        return await behavior(argument, context), True
    except Exception as e:  # pylint: disable=broad-except
        with state.condition:
            if e not in state.rpc_errors:
                details = 'Exception calling application: '.format(e)
                _server.logging.exception(details)
                _server._abort(state, rpc_event.operation_call,
                       _server.cygrpc.StatusCode.unknown, _server._common.encode(details))
        return None, False

async def _take_response_from_response_iterator(rpc_event, state, response_iterator):
    try:
        return await response_iterator.__anext__(), True
    except StopAsyncIteration:
        return None, True
    except Exception as e:  # pylint: disable=broad-except
        with state.condition:
            if e not in state.rpc_errors:
                details = 'Exception iterating responses: '.format(e)
                _server.logging.exception(details)
                _server._abort(state, rpc_event.operation_call,
                       _server.cygrpc.StatusCode.unknown, _server._common.encode(details))
        return None, False

async def _unary_response_in_pool(rpc_event, state, behavior, argument_thunk,
                                  request_deserializer, response_serializer):
    argument = argument_thunk()
    if argument is not None:
        response, proceed = await _call_behavior(rpc_event, state, behavior,
                                                 argument, request_deserializer)
        if proceed:
            serialized_response = _server._serialize_response(
                rpc_event, state, response, response_serializer)
            if serialized_response is not None:
                _server._status(rpc_event, state, serialized_response)

async def _stream_response_in_pool(rpc_event, state, behavior, argument_thunk,
                                   request_deserializer, response_serializer):
    argument = argument_thunk()
    if argument is not None:
        # Notice this calls the normal `_call_behavior` not the awaitable version.
        response_iterator, proceed = _server._call_behavior(
            rpc_event, state, behavior, argument, request_deserializer)
        if proceed:
            while True:
                response, proceed = await _take_response_from_response_iterator(
                    rpc_event, state, response_iterator)
                if proceed:
                    if response is None:
                        _server._status(rpc_event, state, None)
                        break
                    else:
                        serialized_response = _server._serialize_response(
                            rpc_event, state, response, response_serializer)
                        print(response)
                        if serialized_response is not None:
                            print("Serialized Correctly")
                            proceed = _server._send_response(rpc_event, state,
                                                     serialized_response)
                            if not proceed:
                                break
                        else:
                            break
                else:
                    break

_server._unary_response_in_pool = _unary_response_in_pool
_server._stream_response_in_pool = _stream_response_in_pool


if __name__ == '__main__':
    server = grpc.server(AsyncioExecutor())
    # Add Servicer and Start Server Here

enlace al original:
https://gist.github.com/seglberg/0b4487b57b4fd425c56ad72aba9971be

Se puede hacer de forma asincrónica si su llamada a res.get se puede hacer de forma asincrónica (si se define con el async palabra clave).

Tiempo grpc.server dice que requiere un futures.ThreadPoolExecutor, en realidad funcionará con cualquier futures.Executor que llama a los comportamientos que se le envían en algún subproceso distinto al que se transmitieron. ¿Pasarías a grpc.server a futures.Executor implementado por usted que solo usó un hilo para realizar cuatrocientas (o más) llamadas simultáneas a EventReporting.ReportEvent, su servidor debe evitar el tipo de bloqueo que describe.

Reseñas y puntuaciones del artículo

Si te animas, eres capaz de dejar un artículo acerca de qué te ha impresionado de este enunciado.

¡Haz clic para puntuar esta entrada!
(Votos: 0 Promedio: 0)



Utiliza Nuestro Buscador

Deja una respuesta

Tu dirección de correo electrónico no será publicada. Los campos obligatorios están marcados con *