Multithreaded approach to process SQS item Queue

2
votes

In this scenerio, I have to Poll AWS SQS messages from a queue, each async request can fetch upto 10 sqs items/messages. Once I Poll the items, Then I have to process those items on a kubernetes pod. Item processing includes getting response from few API calls, it may take some time & then saving the item to DB & S3. I did some R&D & reach on following conclusion

  1. To use consumer producer model, 1 thread will poll items & another thread will process the item or to use multi-threading for item processing
  2. Maintain a data structure that will containes sqs polled items ready for processing, DS could be Blocking collection or Concurrent queue
  3. Using Task Parellel Library for threadpooling & in item processing.
  4. Channels can be used

My Queries

  1. What would be best approach to achieve best performance or increase TPS.
  2. Can/Should I use data flow TPL
  3. Multi threaded or single threaded with asyn tasks
2
c#multithreading.net-coretask-parallel-librarytpl-dataflow
How many messages per second are you anticipating? You could do all kind of magic tricks but if the rate is, say 1 msg per second, you can make other choices. – Peter Bons
We want to achieve 500 TPS, I mentioned it incorrectly that processing through API call takes seconds, but it takes millisecond get result for an API call – panky sharma
Do you need to process them in order? – Peter Bons
No that's not required – panky sharma
Could you please share with some code that shows how do you imagine a single threaded processing? – Peter Csala

2 Answers

0
votes

I'm not familiar with Kubernetes but there are many things to consider when maximising throughput.

All the things which you have mentioned is IO bound not CPU bound. So, using TPL is overcomplicating the design for marginal benefit. See: https://docs.microsoft.com/en-us/dotnet/csharp/async#recognize-cpu-bound-and-io-bound-work

Your Kubernetes pods are likely to have network limitations. For example, with Azure Function Apps on Consumption Plans is limited to 1,200 outbound connections. Other services will have some defined limits, too. https://docs.microsoft.com/en-us/azure/azure-functions/manage-connections?tabs=csharp#connection-limit. Due to the nature of your work, it is likely that you will reach these limits before you need to process IO work on multiple threads.

You may also need to consider limits of the services which you are dependent on and ensure they are able to handle the throughput.

You may want to consider using Semaphores to limit the number of active connections to satisfy both your infrastructure and external dependency limits https://docs.microsoft.com/en-us/dotnet/api/system.threading.semaphoreslim?view=net-5.0

That being said, 500 messages per second is a realistic amount. To improve it further, you can look at having multiple processes with independent resource limitations processing the queue.

0
votes

  1. What would be best approach to achieve best performance or increase TPS:

    • I would consider using pods autoscaling based on your SQS queue size. Ready to go solutions: KEDA, AWS CW metrics;

  2. Can/Should I use data flow TPL:

  3. Multi threaded or single threaded with async tasks:

    • Since you are running I/O operations (S3 uploads, DB queries), your thread kicks off the process and goes back to the pool, waiting for it's turn to proceed the task. While it's free, it can be utilized for the other process to kick off. That said, you don't need two threads to do the job, but I would let the task scheduler to decide. I would go like:
var dbSaveTask = KickOffTheDbSave();
var s3SaveTask = KickOffTheS3Save();
await Task.WhenAll(dbSaveTask, s3SaveTask);

I am not fully aware of your processes so these are not recommendations but rather things to consider:

  • Since you're working with AWS, and are saving files into S3, you could attach a lambda triggered by S3 upload event to then store the item into the database. You can think of it as of certain transaction. You won't have a record in the database until you have the document. Also, it will segregate the processes to follow SOLID principles;
  • Consider using Bulkhead Isolation policy for recourses limiting.