Operação assíncrona da NDB

Ao otimizar o desempenho de uma aplicação, considere a respetiva utilização da NDB. Por exemplo, se uma aplicação ler um valor que não está na cache, essa leitura demora algum tempo. Pode acelerar a sua aplicação executando ações do Datastore em paralelo com outras ações ou executando algumas ações do Datastore em paralelo entre si.

A biblioteca cliente NDB oferece muitas funções assíncronas ("async"). Cada uma destas funções permite que uma aplicação envie um pedido para o Datastore. A função devolve imediatamente um objeto Future. A aplicação pode fazer outras coisas enquanto o Datastore processa o pedido. Depois de o Datastore processar o pedido, a aplicação pode obter os resultados do objeto Future.

Introdução

Suponhamos que um dos controladores de pedidos da sua aplicação tem de usar o NDB para escrever algo, talvez para registar o pedido. Também tem de realizar outras operações NDB, talvez para obter alguns dados. 

class MyRequestHandler(webapp2.RequestHandler):
    def get(self):
        acct = Account.get_by_id(users.get_current_user().user_id())
        acct.view_counter += 1
        acct.put()

        # ...read something else from Datastore...

        self.response.out.write('Content of the page')

Ao substituir a chamada para put() por uma chamada para o respetivo equivalente assíncrono put_async(), a aplicação pode fazer outras coisas imediatamente em vez de bloquear em put(). 

class MyRequestHandler(webapp2.RequestHandler):
    def get(self):
        acct = Account.get_by_id(users.get_current_user().user_id())
        acct.view_counter += 1
        future = acct.put_async()

        # ...read something else from Datastore...

        self.response.out.write('Content of the page')
        future.get_result()

Isto permite que as outras funções NDB e a renderização de modelos ocorram enquanto o Datastore escreve os dados. A aplicação não bloqueia no Datastore até receber dados do Datastore.

Neste exemplo, é um pouco ridículo chamar future.get_result: a aplicação nunca usa o resultado do NDB. Esse código está apenas lá para garantir que o controlador de pedidos não termina antes de o NDB put terminar. Se o controlador de pedidos terminar demasiado cedo, a colocação pode nunca acontecer. Para sua conveniência, pode decorar o controlador de pedidos com @ndb.toplevel. Isto indica ao controlador que não deve sair até que os pedidos assíncronos tenham terminado. Isto, por sua vez, permite-lhe enviar o pedido e não se preocupar com o resultado.

Pode especificar um WSGIApplication inteiro como ndb.toplevel. Isto garante que cada um dos controladores de WSGIApplication aguarda todos os pedidos assíncronos antes de ser devolvido. (Não "toplevel" todos os controladores de WSGIApplication.) 


app = ndb.toplevel(webapp2.WSGIApplication([('/', MyRequestHandler)]))

Usar uma aplicação toplevel é mais conveniente do que todas as respetivas funções de controlador. No entanto, se um método de controlador usar yield, esse método continua a ter de ser incluído noutro decorador, @ndb.synctasklet; caso contrário, deixa de ser executado no yield e não termina. 

class MyRequestHandler(webapp2.RequestHandler):
    @ndb.toplevel
    def get(self):
        acct = Account.get_by_id(users.get_current_user().user_id())
        acct.view_counter += 1
        acct.put_async()  # Ignoring the Future this returns

        # ...read something else from Datastore...

        self.response.out.write('Content of the page')

Usar APIs assíncronas e Futures

Quase todas as funções síncronas do NDB têm uma contrapartida _async. Por exemplo, put() tem put_async(). Os argumentos da função assíncrona são sempre os mesmos que os da versão síncrona. O valor devolvido de um método assíncrono é sempre um Future ou (para funções "multi") uma lista de Futures.

Um Future é um objeto que mantém o estado de uma operação que foi iniciada, mas que ainda pode não ter sido concluída. Todas as APIs assíncronas devolvem um ou mais Futures. Pode chamar a função Future's get_result() para lhe pedir o resultado da respetiva operação; o Future é então bloqueado, se necessário, até o resultado estar disponível, e, em seguida, é-lhe apresentado. get_result() devolve o valor que seria devolvido pela versão síncrona da API.

Nota: se usou Futures noutras linguagens de programação, pode pensar que pode usar um Future como resultado diretamente. Isso não funciona aqui. Esses idiomas usam futuros implícitos; o NDB usa futuros explícitos. Ligue para get_result() para obter o resultado de um NDB Future.

E se a operação gerar uma exceção? Isso depende do momento em que a exceção ocorre. Se o NDB detetar um problema ao fazer um pedido (talvez um argumento do tipo errado), o método _async() gera uma exceção. No entanto, se a exceção for detetada, por exemplo, pelo servidor do Datastore, o método _async() devolve um Future e a exceção é apresentada quando a sua aplicação chama o respetivo get_result(). Não se preocupe demasiado com isto, pois tudo acaba por se comportar de forma bastante natural. Talvez a maior diferença seja que, se for impresso um rastreio, verá algumas partes do mecanismo assíncrono de baixo nível expostas.

Por exemplo, suponhamos que está a escrever uma aplicação de livro de visitas. Se o utilizador tiver sessão iniciada, quer apresentar uma página que mostre as publicações mais recentes do livro de visitas. Esta página também deve mostrar o nickname do utilizador. A aplicação precisa de dois tipos de informações: as informações da conta do utilizador com sessão iniciada e os conteúdos das publicações no livro de visitas. A versão "síncrona" desta aplicação pode ter o seguinte aspeto: 

uid = users.get_current_user().user_id()
acct = Account.get_by_id(uid)  # I/O action 1
qry = Guestbook.query().order(-Guestbook.post_date)
recent_entries = qry.fetch(10)  # I/O action 2

# ...render HTML based on this data...
self.response.out.write('<html><body>{}</body></html>'.format(''.join(
    '<p>{}</p>'.format(entry.content) for entry in recent_entries)))

Existem duas ações de E/S independentes aqui: obter a entidade Account e obter entidades Guestbook recentes. Com a API síncrona, estas ações ocorrem uma após a outra; aguardamos a receção das informações da conta antes de obter as entidades do livro de visitas. No entanto, a aplicação não precisa das informações da conta imediatamente. Podemos tirar partido desta situação e usar APIs assíncronas: 

uid = users.get_current_user().user_id()
acct_future = Account.get_by_id_async(uid)  # Start I/O action #1
qry = Guestbook.query().order(-Guestbook.post_date)
recent_entries_future = qry.fetch_async(10)  # Start I/O action #2
acct = acct_future.get_result()  # Complete #1
recent_entries = recent_entries_future.get_result()  # Complete #2

# ...render HTML based on this data...
self.response.out.write('<html><body>{}</body></html>'.format(''.join(
    '<p>{}</p>'.format(entry.content) for entry in recent_entries)))

Esta versão do código cria primeiro dois Futures (acct_future e recent_entries_future) e, em seguida, aguarda por eles. O servidor funciona em ambos os pedidos em paralelo. Cada chamada de função _async() cria um objeto Future e envia um pedido para o servidor do Datastore. O servidor pode começar a trabalhar no pedido imediatamente. As respostas do servidor podem ser devolvidas em qualquer ordem arbitrária. O objeto Future associa as respostas aos respetivos pedidos correspondentes.

Os pedidos síncronos não se sobrepõem, mas os assíncronos podem sobrepor-se.
Pedidos síncronos vs. assíncronos

O tempo total (real) gasto na versão assíncrona é aproximadamente igual ao tempo máximo nas operações. O tempo total gasto na versão síncrona excede a soma dos tempos de operação. Se puder executar mais operações em paralelo, as operações assíncronas são mais úteis.

Para ver quanto tempo demoram as consultas da sua aplicação ou quantas operações de E/S são realizadas por pedido, considere usar o Appstats. Esta ferramenta pode apresentar gráficos semelhantes ao desenho acima com base na instrumentação de uma app em direto.

Usar minitarefas

Um tasklet NDB é um elemento de código que pode ser executado em simultâneo com outro código. Se escrever um tasklet, a sua aplicação pode usá-lo de forma semelhante a como usa uma função NDB assíncrona: chama o tasklet, que devolve um Future; mais tarde, chamar o método get_result() do Future obtém o resultado.

Os tasklets são uma forma de escrever funções concorrentes sem threads. Os tasklets são executados por um ciclo de eventos e podem suspender-se a si próprios, bloqueando a E/S ou alguma outra operação através de uma declaração de rendimento. A noção de uma operação de bloqueio é abstraída na classe Future, mas um tasklet também pode yield um RPC para aguardar a conclusão desse RPC. Quando o tasklet tem um resultado, é uma exceção raisendb.Return; o NDB associa então o resultado ao yieldFuture com script prévio.

Quando escreve um tasklet NDB, usa yield e raise de uma forma invulgar. Assim, se procurar exemplos de como usar estas, provavelmente não vai encontrar código como um tasklet NDB.

Para transformar uma função num tasklet NDB:

  • decorar a função com @ndb.tasklet,
  • substituir todas as chamadas de armazenamento de dados síncronas por yields de chamadas de armazenamento de dados assíncronas,
  • fazer com que a função "devolva" o respetivo valor de retorno com raise ndb.Return(retval) (não é necessário se a função não devolver nada).

Uma aplicação pode usar tasklets para um controlo mais preciso sobre as APIs assíncronas. Por exemplo, considere o seguinte esquema: 

class Account(ndb.Model):
    email = ndb.StringProperty()
    nickname = ndb.StringProperty()

    def nick(self):
        return self.nickname or self.email  # Whichever is non-empty
... 
class Message(ndb.Model):
    text = ndb.StringProperty()
    when = ndb.DateTimeProperty(auto_now_add=True)
    author = ndb.KeyProperty(kind=Account)  # references Account

Ao apresentar uma mensagem, faz sentido mostrar o pseudónimo do autor. A forma "síncrona" de obter os dados para apresentar uma lista de mensagens pode ter o seguinte aspeto: 

qry = Message.query().order(-Message.when)
for msg in qry.fetch(20):
    acct = msg.author.get()
    self.response.out.write(
        '<p>On {}, {} wrote:'.format(msg.when, acct.nick()))
    self.response.out.write('<p>{}'.format(msg.text))

Infelizmente, esta abordagem é ineficiente. Se o tivesse analisado no Appstats, veria que os pedidos "Get" estão em série. Pode ver o seguinte padrão de "escada".

As obtenções síncronas ocorrem em série
As "obtenções" síncronas ocorrem em série.

Esta parte do programa seria mais rápida se essas "obtenções" pudessem sobrepor-se. Pode reescrever o código para usar get_async, mas é difícil controlar que pedidos e mensagens assíncronos pertencem uns aos outros.

A aplicação pode definir a sua própria função "async" transformando-a num tasklet. Isto permite-lhe organizar o código de uma forma menos confusa.

Além disso, em vez de usar acct = key.get() ou acct = key.get_async().get_result(), a função deve usar acct = yield key.get_async(). Este yield indica ao NDB que este é um bom local para suspender este tasklet e permitir que outros tasklets sejam executados.

Decorar uma função geradora com @ndb.tasklet faz com que a função devolva um Future em vez de um objeto gerador. No tasklet, qualquer yield de um Future aguarda e devolve o resultado do Future.

Por exemplo: 

@ndb.tasklet
def callback(msg):
    acct = yield msg.author.get_async()
    raise ndb.Return('On {}, {} wrote:\n{}'.format(
        msg.when, acct.nick(), msg.text))

qry = Message.query().order(-Message.when)
outputs = qry.map(callback, limit=20)
for output in outputs:
    self.response.out.write('<p>{}</p>'.format(output))

Tenha em atenção que, embora get_async() devolva um Future, a framework de tasklets faz com que a expressão yield devolva o resultado de Future à variável acct.

O map() chama callback() várias vezes. Mas o yield ..._async() em callback() permite que o agendador do NDB envie muitos pedidos assíncronos antes de aguardar que qualquer um deles termine.

Sobreposição de &quot;Gets&quot; assíncronos
"Gets" assíncronos sobrepostos

Se analisar esta situação no Appstats, pode ficar surpreendido ao ver que estas várias obtenções não se sobrepõem apenas, mas são todas processadas no mesmo pedido. O NDB implementa um "autobatcher". O autobatcher agrupa vários pedidos num único RPC em lote para o servidor; faz isto de tal forma que, desde que haja mais trabalho a fazer (pode ser executado outro callback), recolhe chaves. Assim que um dos resultados for necessário, o autobatcher envia o RPC em lote. Ao contrário da maioria dos pedidos, as consultas não são "agrupadas".

Quando um tasklet é executado, recebe o respetivo espaço de nomes predefinido do que era predefinido quando o tasklet foi gerado ou do que o tasklet alterou durante a execução. Por outras palavras, o espaço de nomes predefinido não está associado nem armazenado no Context, e a alteração do espaço de nomes predefinido num tasklet não afeta o espaço de nomes predefinido noutros tasklets, exceto nos gerados por ele.

Tasklets, consultas paralelas e rendimento paralelo

Pode usar tasklets para que várias consultas obtenham registos em simultâneo. Por exemplo, suponhamos que a sua aplicação tem uma página que apresenta o conteúdo de um carrinho de compras e uma lista de ofertas especiais. O esquema pode ter o seguinte aspeto: 

class Account(ndb.Model):
    pass


class InventoryItem(ndb.Model):
    name = ndb.StringProperty()


class CartItem(ndb.Model):
    account = ndb.KeyProperty(kind=Account)
    inventory = ndb.KeyProperty(kind=InventoryItem)
    quantity = ndb.IntegerProperty()


class SpecialOffer(ndb.Model):
    inventory = ndb.KeyProperty(kind=InventoryItem)

Uma função "síncrona" que obtém artigos do carrinho de compras e ofertas especiais pode ter o seguinte aspeto: 

def get_cart_plus_offers(acct):
    cart = CartItem.query(CartItem.account == acct.key).fetch()
    offers = SpecialOffer.query().fetch(10)
    ndb.get_multi([item.inventory for item in cart] +
                  [offer.inventory for offer in offers])
    return cart, offers

Este exemplo usa consultas para obter listas de artigos do carrinho e ofertas. Em seguida, obtém detalhes sobre os artigos do inventário com get_multi(). (Esta função não usa diretamente o valor de retorno de get_multi(). Chama get_multi() para obter todos os detalhes do inventário para a cache, para que possam ser lidos rapidamente mais tarde.) get_multi combina muitos Gets num único pedido. No entanto, as obtenções de consultas ocorrem uma após a outra. Para que essas obtenções ocorram em simultâneo, sobreponha as duas consultas: 

def get_cart_plus_offers_async(acct):
    cart_future = CartItem.query(CartItem.account == acct.key).fetch_async()
    offers_future = SpecialOffer.query().fetch_async(10)
    cart = cart_future.get_result()
    offers = offers_future.get_result()
    ndb.get_multi([item.inventory for item in cart] +
                  [offer.inventory for offer in offers])
    return cart, offers

A chamada get_multi() continua a ser separada: depende dos resultados da consulta, pelo que não a pode combinar com as consultas.

Suponhamos que esta aplicação precisa, por vezes, do carrinho, por vezes, das ofertas e, por vezes, de ambos. Quer organizar o seu código para que exista uma função para obter o carrinho e uma função para obter as ofertas. Se a sua aplicação chamar estas funções em conjunto, idealmente, as respetivas consultas podem "sobrepor-se". Para tal, transforme estas funções em tasklets: 

@ndb.tasklet
def get_cart_tasklet(acct):
    cart = yield CartItem.query(CartItem.account == acct.key).fetch_async()
    yield ndb.get_multi_async([item.inventory for item in cart])
    raise ndb.Return(cart)


@ndb.tasklet
def get_offers_tasklet(acct):
    offers = yield SpecialOffer.query().fetch_async(10)
    yield ndb.get_multi_async([offer.inventory for offer in offers])
    raise ndb.Return(offers)


@ndb.tasklet
def get_cart_plus_offers_tasklet(acct):
    cart, offers = yield get_cart_tasklet(acct), get_offers_tasklet(acct)
    raise ndb.Return((cart, offers))

Isso yield x, y é importante, mas é fácil de ignorar. Se fossem duas declarações yield separadas, ocorreriam em série. No entanto, yielding uma tupla de tasklets é um rendimento paralelo: os tasklets podem ser executados em paralelo e o yield aguarda que todos terminem e devolve os resultados. (Em algumas linguagens de programação, isto é conhecido como uma barreira.)

Se transformar um fragmento de código num tasklet, é provável que queira fazer mais em breve. Se reparar num código "síncrono" que possa ser executado em paralelo com um tasklet, é provavelmente uma boa ideia torná-lo também um tasklet. Em seguida, pode paralelizar com um yield paralelo.

Se escrever uma função de pedido (uma função de pedido webapp2, uma função de visualização Django, etc.) para ser um tasklet, não vai fazer o que quer: produz, mas depois para de ser executada. Nesta situação, quer decorar a função com @ndb.synctasklet. @ndb.synctasklet é semelhante a @ndb.tasklet, mas foi alterado para chamar get_result() no tasklet. Isto transforma o seu pequeno programa numa função que devolve o resultado da forma habitual.

Consultar iteradores em Tasklets

Para iterar os resultados da consulta num tasklet, use o seguinte padrão: 

qry = Model.query()
qit = qry.iter()
while (yield qit.has_next_async()):
    entity = qit.next()
    # Do something with entity
    if is_the_entity_i_want(entity):
        raise ndb.Return(entity)

Esta é a alternativa compatível com tasklets do seguinte: 

# DO NOT DO THIS IN A TASKLET
qry = Model.query()
for entity in qry:
    # Do something with entity
    if is_the_entity_i_want(entity):
        raise ndb.Return(entity)

As três linhas a negrito na primeira versão são o equivalente compatível com tasklets da única linha a negrito na segunda versão. Os tasklets só podem ser suspensos ao nível de uma yieldpalavra-chave. O ciclo for sem yield não permite a execução de outros tasklets.

Pode perguntar-se por que motivo este código usa um iterador de consultas em vez de obter todas as entidades através de qry.fetch_async(). A aplicação pode ter tantas entidades que não cabem na RAM. Talvez esteja à procura de uma entidade e possa parar de iterar assim que a encontrar. No entanto, não pode expressar os seus critérios de pesquisa apenas com a linguagem de consulta. Pode usar um iterador para carregar entidades a verificar e, em seguida, sair do ciclo quando encontrar o que quer.

Async Urlfetch com NDB

Um NDB Context tem uma função urlfetch() assíncrona que funciona bem em paralelo com os tasklets do NDB, por exemplo: 

@ndb.tasklet
def get_google():
    context = ndb.get_context()
    result = yield context.urlfetch("http://www.google.com/")
    if result.status_code == 200:
        raise ndb.Return(result.content)

O serviço de obtenção de URL tem a sua própria API de pedidos assíncronos. Não há problema, mas nem sempre é fácil de usar com tasklets NDB.

Usar transações assíncronas

As transações também podem ser feitas de forma assíncrona. Pode transmitir uma função existente para ndb.transaction_async() ou usar o decorador @ndb.transactional_async. Tal como as outras funções assíncronas, esta devolve um NDB Future: 

@ndb.transactional_async
def update_counter(counter_key):
    counter = counter_key.get()
    counter.value += 1
    counter.put()

As transações também funcionam com tasklets. Por exemplo, podemos alterar o nosso código update_counter para yield enquanto aguardamos RPCs de bloqueio: 

@ndb.transactional_tasklet
def update_counter(counter_key):
    counter = yield counter_key.get_async()
    counter.value += 1
    yield counter.put_async()

Usar Future.wait_any()

Por vezes, quer fazer vários pedidos assíncronos e devolvê-los sempre que o primeiro for concluído. Pode fazê-lo através do método de classe ndb.Future.wait_any(): 

def get_first_ready():
    urls = ["http://www.google.com/", "http://www.blogspot.com/"]
    context = ndb.get_context()
    futures = [context.urlfetch(url) for url in urls]
    first_future = ndb.Future.wait_any(futures)
    return first_future.get_result().content

Infelizmente, não existe uma forma conveniente de transformar isto num tasklet; um yield paralelo aguarda a conclusão de todos os Futures, incluindo aqueles pelos quais não quer esperar.