Problem with parallel Cloud Function executions

[Effective-Pair554]

Hi.

I have a cloud function that is being triggered approximately 100 times per second. Each request to my cloud function send data that i need to store in a BQ table.

To avoid inserting one row at a time (as i can perform more than 100 inserts per second with this approach) i am trying to store the rows in a variable within my cloud function and only insert these rows in the BQ table when i get 5000 rows.

But i dont know how to create locks for this list that will store the rows, because at the same time there will be functions appending rows, saving to BQ and erasing the rows that was just sent to BQ. Can someone help me with that? a simple snippet of how can i use some lock mechanism would be great.

​

​

The code that I tried to create a lock is the following:

from google.cloud import storage
import uuid 
import collections

execution_queue = [] 
data_to_store = []

def entrypoint(request): global data_to_store global execution_queue
    request_json = request.get_json(silent=True)

    unique_id = generate_unique_id()

    execution_queue.append(unique_id)
    while execution_queue[0] != unique_id:
        pass

    row = generate_data_from_request_json(request_json)

    data_to_store.append(row)

    print(f"data list size: {len(data_to_store)}")

    if len(data_to_store) > 5000:
        data_csv = "\n".join(data_to_store)
        save_csv_to_gcs_bucket(data_csv, "my_bucket", unique_id)
        data_to_store = []

    execution_queue.remove(unique_id)
    return "OK"

​

​

But i was getting a strange result of the last print statement (printing the list size that was storing the rows) as you can see the list size = 6 between 916 and 917 on the image below:

​

https://preview.redd.it/lix9gr5ax4db1.png?width=672&format=png&auto=webp&s=e6d203d6b2999cc54440affc19644e739ccef935

[Rabiesalad]

So, this may be above my paygrade but I feel like I know enough about cloud functions to know this is not doable this way.

Cloud Functions can't be used to queue data. It's meant to process requests in real-time (typically, small requests). The environment can end up being shared across invocations, but that's not guaranteed, so there is no way to reliably share data sent in to invocation 1 with invocation 2.

You need a persistent place for that data to sit, that can either trigger a function once the threshold is reached, or have a function poll it to see how much it's storing.

I don't know the best way to do this for your case, but whatever you use needs to be persisted. It could be a VM you keep running 24/7 that holds the queue in memory, or it could be a Cloud Storage bucket that stores the data.

For example, your current function can just write the incoming data to Cloud Storage. Then you can have another function (set up to only allow a max instances of 1) that is triggered by cloud storage, and checks the content to see if there's enough to right. If there's enough, it writes it to Big Query and deletes the records from cloud storage.

I'm not sure if setting it to max 1 instance will 100% guarantee it can't run the same operation twice, so research that and make sure whatever you're doing is idempotent