Using Parallel to Enqueue 10,000 Tasks from Rails to Cloud Tasks: A Performance Record

I encountered a situation where I needed to extract all records (on the scale of tens of thousands) from a table stored in a database and perform processing on each one. For example... a case like creating HTML emails from user profile text stored in the DB and sending them out one by one.

From a cloud architecture perspective, utilizing a queue such as Google Cloud Tasks is convenient. As long as you can push data into the queue, Cloud Tasks allows you to finely tune the throughput. Let's consider a configuration roughly like this:

From the standpoint of idempotency and scaling, it is safer to design the system so that one Cloud Task outputs a specific result. When dealing with all DB data, you might feel tempted to pack a batch of arrays into a single task to send emails, but we will follow the policy of sending one email per task.

As long as you push them into Cloud Tasks, load adjustment is flexible

A bit out of order, but let's talk about what happens after all task data has been pushed into Cloud Tasks. Naturally, the email sending app has a server, and its capacity is limited. There are also situations where you must consider rate limits when using a SaaS. Often, you want to proceed slowly without putting too much load on the system since this kind of bulk processing can afford to take some time. In Cloud Tasks, you can adjust the number of concurrent executions and dispatches per second (similar to a processing wait queue) according to such situations. This blog post is very helpful, so please check it out if you want to know more details.

https://zenn.dev/nananaoto/articles/c62d63223e669af63c80

The Challenge: What about the Enqueueing side?

Now, what I want to mention in this article is the task enqueueing side. When enqueuing a large amount of data, the resources that must be considered are "time" and "computing resources." If it's not particularly urgent, you might think, "since it's fine if it takes some time, I'll just enqueue everything slowly..." So, I used an enqueueing logic composed of the following pseudo-code.

  def enqueue
    tasks = CloudTasks.new
    User.all.find_each do |user|
        create_params = {
          queue: "send-mail",
          target_path: "/mail/sender",
          payload: {
            user_id: user.id,
            user_name: user.name,
            email: user.email
          }
        }
        tasks.create(**create_params)
      end
    end
  end

Since it uses find_each, it selects 1,000 records at a time and executes repeatedly. When I ran this code with 10,000 dummy users, it took 15 minutes to enqueue everything to Cloud Tasks. ^[1]

You could argue that it's fine since it's not urgent, but the maximum request execution time for the Google App Engine flexible environment is 60 minutes. If it exceeds this, the process will time out and the enqueueing will be interrupted. At this rate, there's a risk of hitting this limit as the number of users increases in the future. Improvement is necessary.

Improvement Idea 1️⃣ for the Enqueueing Side: Make it a Rake Task

This strategy involves enqueuing via a Rake task instead of an HTTPS request. While it's a very valid option, I want to keep Rake tasks as a last resort for the following two reasons:

• As the frequency of deploying apps as containers increases, adopting a Rake task requires considering whether to enable SSH or set up a dedicated container for Rake tasks.
• It becomes harder to delegate the sequence of starting the process and confirming completion to someone other than a developer.^[2]

In short, because the things to think about and maintain tend to increase, I'll set this candidate aside for now.

Improvement Idea 2️⃣ for the Enqueueing Side: Adopting the Dual-Injection Pattern

Looking at the Cloud Tasks documentation, it also mentions enqueueing.

When you need to add a large number of tasks, such as millions or billions, a dual-injection pattern is useful. Instead of creating tasks from a single job, use an injector queue. Each task added to the injector queue fans out to add 100 tasks to the target queue or queue group.
https://cloud.google.com/tasks/docs/manage-cloud-task-scaling?hl=en#large-scalebatch_task_enqueues

In other words, it suggests preparing a Cloud Tasks queue specifically for pushing individual tasks into Cloud Tasks.

I understand the concept, but it feels too large-scale and difficult for this case. Since we are dealing with tens of thousands of records, I don't want to go that far. I'll think of another method. However, this approach is attractive because it can handle data on the scale of a million records as long as there is enough time. It's worth considering if the data grows explosively.

Improvement Idea 3️⃣ for the Enqueueing Side: Parallel Enqueueing

Based on advice from a colleague, I'll consider this option this time. I decided to try using the Parallel gem. However, since it was hard to predict how much it would improve, I recorded some actual measurements, which I'll share here.

Introduction of Parallel

https://github.com/grosser/parallel

The introduction was simple and quick. I rewrote the part that enqueues to Cloud Tasks as follows:

  BULK_ENQUEUE_TASKS_THREADS = ???
  
  def enqueue
    tasks = CloudTasks.new
+    User.all.select(:id).find_in_batches do |items| # ①
+      Parallel.each(items, in_threads: BULK_ENQUEUE_TASKS_THREADS) do |item| # ②
        create_params = {
          queue: "send-mail",
          target_path: "/mail/sender",
          payload: {
            user_id: user.id,
            user_name: user.name,
            email: user.email
          }
        }
        tasks.create(**create_params)
      end
    end
  end

1. Since Parallel takes an array and performs parallel processing, I changed find_each to find_in_batches.
2. Specify the number of threads. I want to determine this value.

Execution Environment

Item Name	Value
Google App Engine Flexible Instance	cpu:1, memory_gb: 2, disk_size_gb: 10
Rails Version	6.1
Parallel Version	1.20.1
RAILS_MAX_THREADS	5

Actual Measurement

The processing speed itself varies greatly depending on your environment and conditions. Please use the measured values as a reference only.

Now, let's measure while changing the number of logical threads in Parallel.

Time required to enqueue 10,000 tasks (seconds)

Logical Threads	1st	2nd	3rd	4th	5th	Median (sec)
Without Parallel	900.00					900.00 (15 min)
5	159.07	159.12	160.66	160.97	161.62	160.66
20	40.45	40.45	40.45	40.66	40.15	40.45
50	17.21	17.21	18.52	18.52	18.52	18.52
100	11.67	11.86	11.76	11.39	12.17	11.76
500	11.82	10.90	11.73	12.50	10.89	11.73
1000	13.50	13.50	12.78	12.98	12.59	12.98

It has become incredibly fast. Honestly, I was surprised. I was hoping that 15 minutes would somehow be cut in half, but it unexpectedly finished in about 12 seconds. It seems that 100 to 500 logical threads is the upper limit for throughput, and beyond that, the overhead of thread creation starts to outweigh the benefits. For this project, I set the number of threads to 100.

Is the Cloud Tasks side okay?

You might feel anxious, wondering if the tasks are being enqueued without any omissions. They are being fully enqueued without any issues. No matter how much the number of threads is increased, 10,000 records are registered. While the efficiency gained from multi-threading is impressive, Cloud Tasks, which remains completely unfazed by it, is also remarkable. I looked for documentation regarding rate limits and similar constraints, but I couldn't find any specific upper limits for the enqueueing side. If anyone has any information, please let me know. ^[3]

Status where 10,000 tasks have been enqueued to Cloud Tasks

Conclusion: By adopting Parallel, the enqueueing time was reduced from 900 seconds to 12 seconds

By simple calculation, if it takes 15 minutes to enqueue 10,000 items, it would hit a timeout when it reaches 40,000 items. By utilizing multi-threading, we were able to shorten the time to 12 seconds, so we likely won't have to worry about enqueueing timeouts for a while. While the trend in cloud and container environments is horizontal scaling of load, this experience was a good reminder that it's also worth stepping back and looking at other options. I think it's interesting to try using machine resources to their limit with multi-threading or multi-processing before figuring out how to partition the data.

Reference

https://www.xmisao.com/2018/07/22/how-to-use-ruby-parallel-gem.html

https://zenn.dev/nananaoto/articles/bd1584c77e46f128a41a

脚注

1. It would be great if Cloud Tasks had a batch_send style API, but it doesn't seem to exist so far

2. If you make it possible to interact via HTTPS, you can send requests from an admin panel or integrate it with a chatbot.

3. I was informed in the comments. It is reportedly 6,000,000 r/m (100,000 r/s).