Operazione asincrona NDB

Quando ottimizzi le prestazioni di un'applicazione, prendi in considerazione il suo utilizzo di NDB. Ad esempio, se un'applicazione legge un valore che non è nella cache, la lettura richiede un po' di tempo. Potresti essere in grado di velocizzare l'applicazione eseguendo le azioni Datastore in parallelo con altre operazioni o eseguendo alcune azioni Datastore in parallelo tra loro.

La libreria client NDB fornisce molte funzioni asincrone ("async"). Ognuna di queste funzioni consente a un'applicazione di inviare una richiesta a Datastore. La funzione restituisce immediatamente un Future oggetto. L'applicazione può eseguire altre operazioni mentre Datastore gestisce la richiesta. Dopo che Datastore ha gestito la richiesta, l'applicazione può ottenere i risultati dall'oggetto Future.

Utilizzo delle API asincrone e di Future

Quasi ogni funzione NDB sincrona ha una controparte _async. Ad esempio, put() ha put_async(). Gli argomenti della funzione asincrona sono sempre gli stessi della versione sincrona. Il valore restituito di un metodo asincrono è sempre un Future o (per le funzioni "multi") un elenco di Futures.

Un Future è un oggetto che mantiene lo stato di un'operazione che è stata avviata ma potrebbe non essere ancora completata; tutte le API asincrone restituiscono uno o più Futures. Puoi chiamare la funzione Future's get_result() per chiedere il risultato dell’operazione; se necessario, Future si blocca finché il risultato non è disponibile, e poi te lo fornisce. get_result() restituisce il valore che verrebbe restituito dalla versione sincrona dell'API.

Nota: se hai utilizzato Future in altri linguaggi di programmazione, potresti pensare di poter utilizzare un Future come risultato direttamente. Non funziona qui. Questi linguaggi utilizzano future implicite; NDB utilizza future esplicite. Chiama get_result() per ottenere il risultato di un Future NDB.

Cosa succede se l'operazione genera un'eccezione? Dipende da quando si verifica l'eccezione. Se NDB rileva un problema durante la creazione di una richiesta (ad esempio un argomento di tipo errato), il _async() metodo genera un'eccezione. Tuttavia, se l'eccezione viene rilevata, ad esempio, dal server Datastore, il metodo _async() restituisce un Future e l'eccezione verrà generata quando l'applicazione chiama get_result(). Non preoccuparti troppo, tutto si comporta in modo abbastanza naturale; forse la differenza più grande è che se viene stampato un traceback, vedrai alcune parti del meccanismo asincrono di basso livello esposte.

Supponiamo, ad esempio, che tu stia scrivendo un'applicazione guestbook. Se l'utente ha eseguito l'accesso, vuoi presentare una pagina che mostra i post più recenti del guestbook. Questa pagina deve mostrare anche il nickname dell'utente. L'applicazione ha bisogno di due tipi di informazioni: i dati dell'account dell'utente che ha eseguito l'accesso e i contenuti dei post del guestbook. La versione "sincrona" di questa applicazione potrebbe essere simile alla seguente: 

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)))

Qui sono presenti due azioni di I/O indipendenti: recuperare l'entità Account e recuperare le entità Guestbook recenti. Utilizzando l'API sincrona, queste operazioni vengono eseguite una dopo l'altra; attendiamo di ricevere i dati dell'account prima di recuperare le entità del guestbook. Tuttavia, l'applicazione non ha bisogno immediatamente dei dati dell'account. Possiamo sfruttare questa situazione e utilizzare le API asincrone: 

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)))

Questa versione del codice crea prima due Futures (acct_future e recent_entries_future), quindi li attende. Il server elabora entrambe le richieste in parallelo. Ogni chiamata di funzione _async() crea un oggetto Future e invia una richiesta al server Datastore. Il server può iniziare a elaborare la richiesta subito. Le risposte del server possono tornare in qualsiasi ordine arbitrario; l'oggetto Future collega le risposte alle loro richieste corrispondenti.

Le richieste sincrone non si sovrappongono, ma quelle asincrone sì.
Richieste sincrone e asincrone

Il tempo totale (reale) trascorso nella versione asincrona è approssimativamente uguale a il tempo massimo tra le operazioni. Il tempo totale trascorso nella versione sincrona supera la somma dei tempi delle operazioni. Se puoi eseguire più operazioni in parallelo, le operazioni asincrone sono più utili.

Utilizzo dei tasklet

Un tasklet NDB è una parte di codice che potrebbe essere eseguita contemporaneamente ad altro codice. Se scrivi un tasklet, la tua applicazione può utilizzarlo in modo simile a una funzione NDB asincrona: chiama il tasklet, che restituisce un Future; in un secondo momento, chiamando il Future's get_result() method si ottiene il risultato.

I tasklet sono un modo per scrivere funzioni simultanee senza thread; i tasklet vengono eseguiti da un loop di eventi e possono sospendersi bloccando l'I/O o un'altra operazione utilizzando un'istruzione yield. La nozione di operazione di blocco viene astratta nella Future classe, ma un tasklet può anche yield un RPC per attendere il completamento di questo RPC. Quando il tasklet ha un risultato, lo raise un' eccezione ndb.Return; NDB associa quindi il risultato al Future precedentemente yield.

Quando scrivi un tasklet NDB, utilizzi yield e raise in modo insolito. Pertanto, se cerchi esempi di come utilizzarli, probabilmente non troverai codice simile a un tasklet NDB.

Per trasformare una funzione in un tasklet NDB:

  • decora la funzione con @ndb.tasklet,
  • sostituisci tutte le chiamate sincrone di datastore con yields di chiamate di datastore asincrono,
  • fai in modo che la funzione "restituisca" il suo valore di ritorno con raise ndb.Return(retval) (non necessario se la funzione non restituisce nulla).

Un'applicazione può utilizzare i tasklet per un controllo più preciso delle API asincrone. Ad esempio, considera lo schema seguente: 

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

Quando visualizzi un messaggio, è opportuno mostrare il nickname dell'autore. Il modo "sincrono" per recuperare i dati per mostrare un elenco di messaggi potrebbe essere simile al seguente: 

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))

Purtroppo, questo approccio è inefficiente. Potresti visualizzare il seguente pattern "a scaletta".

Le operazioni &quot;Get&quot; sincrone si verificano in serie
Le "Gets" sincrone vengono eseguite in serie.

Questa parte del programma sarebbe più veloce se queste "Gets" potessero sovrapporsi. Potresti riscrivere il codice per utilizzare get_async, ma è difficile tenere traccia delle richieste e dei messaggi asincroni che appartengono insieme.

L'applicazione può definire la propria funzione "async" trasformandola in un tasklet. In questo modo, puoi organizzare il codice in modo meno confuso.

Inoltre, anziché utilizzare acct = key.get() o acct = key.get_async().get_result(), la funzione deve utilizzare acct = yield key.get_async(). Questo yield indica a NDB che questo è un buon punto per sospendere questo tasklet e consentire l'esecuzione di altri tasklet.

La decorazione di una funzione generatore con @ndb.tasklet fa in modo che la funzione restituisca un Future anziché un oggetto generatore. All'interno del tasklet, qualsiasi yield di un Future attende e restituisce il Future's risultato.

Ad esempio: 

@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))

Tieni presente che, sebbene get_async() restituisca un Future, il framework del tasklet fa in modo che l'espressione yield restituisca il risultato di Future's alla variabile acct.

Le chiamate map() callback() più volte. Tuttavia, yield ..._async() in callback() consente allo scheduler di NDB di inviare molte richieste asincrone prima di attendere il completamento di una di esse.

Sovrapposizione di &quot;Gets&quot; asincroni
"Gets" asincrone sovrapposte

Se esamini questa situazione in Appstats, potresti rimanere sorpreso di vedere che queste Gets multiple non si sovrappongono, ma vengono eseguite tutte nella stessa richiesta. NDB implementa un "autobatcher". L'autobatcher raggruppa più richieste in un singolo RPC batch al server; lo fa in modo che, finché c'è altro lavoro da fare (un altro callback potrebbe essere eseguito), raccoglie le chiavi. Non appena è necessario uno dei risultati, l'autobatcher invia l'RPC batch. A differenza della maggior parte delle richieste, le query non vengono "raggruppate".

Quando viene eseguito un tasklet, ottiene il suo spazio dei nomi predefinito da quello predefinito quando è stato generato o da quello modificato durante l'esecuzione. In altre parole, lo spazio dei nomi predefinito non è associato o memorizzato nel contesto e la modifica dello spazio dei nomi predefinito in un tasklet non influisce sullo spazio dei nomi predefinito in altri tasklet, ad eccezione di quelli generati da esso.

Tasklet, query parallele, yield parallelo

Puoi utilizzare i tasklet in modo che più query recuperino i record contemporaneamente. Supponiamo, ad esempio, che la tua applicazione abbia una pagina che mostra i contenuti di un carrello degli acquisti e un elenco di offerte speciali. Lo schema potrebbe essere simile al seguente: 

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)

Una funzione "sincrona" che recupera gli articoli del carrello degli acquisti e le offerte speciali potrebbe essere simile alla seguente: 

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

Questo esempio utilizza le query per recuperare gli elenchi di articoli del carrello e offerte, quindi recupera i dettagli degli articoli di inventario con get_multi(). (Questa funzione non utilizza il valore restituito di get_multi() direttamente. Chiama get_multi() per recuperare tutti i dettagli dell'inventario nella cache in modo che possano essere letti rapidamente in un secondo momento.) get_multi combina molte Gets in una singola richiesta. Tuttavia, le query vengono eseguite una dopo l'altra. Per fare in modo che questi recuperi vengano eseguiti contemporaneamente, sovrapponi le due query: 

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

La chiamata get_multi() è ancora separata: dipende dai risultati della query, quindi non puoi combinarla con le query.

Supponiamo che questa applicazione a volte abbia bisogno del carrello, a volte delle offerte, e a volte di entrambi. Vuoi organizzare il codice in modo che ci sia una funzione per recuperare il carrello e una funzione per recuperare le offerte. Se l'applicazione chiama queste funzioni insieme, idealmente le query potrebbero "sovrapporsi." Per farlo, trasforma queste funzioni in tasklet: 

@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))

yield x, y è importante ma facile da trascurare. Se si trattasse di due istruzioni yield separate, verrebbero eseguite in serie. Tuttavia, l'esecuzione di yield di una tupla di tasklet è un yield parallelo: i tasklet possono essere eseguiti in parallelo e yield attende il completamento di tutti e restituisce i risultati. In alcuni linguaggi di programmazione, questa operazione è nota come barriera.

Se trasformi una parte di codice in un tasklet, probabilmente vorrai fare di più a breve. Se noti codice "sincrono" che potrebbe essere eseguito in parallelo con un tasklet, è probabilmente una buona idea trasformarlo in un tasklet. Quindi puoi parallelizzarlo con un yield parallelo.

Se scrivi una funzione di richiesta (una funzione di richiesta webapp2, una funzione di visualizzazione Django e così via) come tasklet, non funzionerà come previsto: esegue yield, ma poi smette di essere eseguita. In questa situazione, devi decorare la funzione con @ndb.synctasklet. @ndb.synctasklet è simile a @ndb.tasklet, ma modificato per chiamare get_result() sul tasklet. In questo modo, il tasklet diventa una funzione che restituisce il risultato nel modo consueto.

Iteratori di query nei tasklet

Per eseguire l'iterazione sui risultati delle query in un tasklet, utilizza il seguente pattern: 

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)

Questo è l'equivalente compatibile con i tasklet di quanto segue: 

# 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)

Le tre righe in grassetto nella prima versione sono l'equivalente compatibile con i tasklet della singola riga in grassetto nella seconda versione. I tasklet possono essere sospesi solo con una parola chiave yield. Il ciclo for senza yield non consente l'esecuzione di altri tasklet.

Potresti chiederti perché questo codice utilizza un iteratore di query anziché recuperare tutte le entità utilizzando qry.fetch_async(). L'applicazione potrebbe avere così tante entità da non rientrare nella RAM. Forse stai cercando un'entità e puoi interrompere l'iterazione una volta che la trovi, ma non puoi esprimere i criteri di ricerca solo con il linguaggio di query. Puoi utilizzare un iteratore per caricare le entità da controllare, quindi uscire dal ciclo quando trovi quello che ti serve.

Urlfetch asincrono con NDB

Un NDB Context ha una funzione asincrona urlfetch() che si parallelizza bene con i tasklet NDB, ad esempio: 

@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)

Il servizio di recupero URL ha una propria API di richiesta asincrona. Va bene, ma non è sempre facile da utilizzare con i tasklet NDB.

Utilizzo delle transazioni asincrone

Le transazioni possono essere eseguite anche in modo asincrono. Puoi passare una funzione esistente a ndb.transaction_async() o utilizzare il decoratore @ndb.transactional_async. Come le altre funzioni asincrone, questa restituirà un Future NDB: 

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

Le transazioni funzionano anche con i tasklet. Ad esempio, potremmo modificare il codice update_counter in modo che esegua yield durante l'attesa degli RPC di blocco: 

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

Utilizzo di Future.wait_any()

A volte vuoi effettuare più richieste asincrone e restituire un valore quando la prima viene completata. Puoi farlo utilizzando il metodo della 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

Purtroppo, non esiste un modo pratico per trasformare questa operazione in un tasklet; un yield parallelo attende il completamento di tutti i Future, inclusi quelli che non vuoi attendere.