ollien
I have to imagine that the answer I’m going to end up with is “you need a different approach”, but I figured I’d ask as a last-ditch before I gave up on something I was working on.
I’m working to optimize an event processor at work. We get a large volume of events from RabbitMQ, which are divvied up into worker processes for processing. These worker processes query Elasticsearch (among other things) to collect the requisite data to complete processing. One problem we ran into is that Elastic’s search queue gets overwhelmed during times of heavy load; the details aren’t important, but the TL;DR is that in Elastic will only allow a certain number of unprocessed queries to build up before it starts rate limiting clients. As a mitigation, I set up a GenServer which would collect query requests from each worker process, and on a fixed interval (or once a high-water mark was reached), perform a single bulk query with Elastic. I then post-process this bulk response and distribute the query results to the requesting processes (using a simple send/receive).
While this strategy worked for mitigating the rate limits I mentioned before, the records we pull from Elastic are sizable chunks of JSON (topping out around 1M, I’d say), so copying the data back to the requesting processes is a non-trivial endeavor. Unfortunately, this produces an unacceptable amount of latency in our event processing.
Herein lies the question: is there a way I can somehow share these chunks of data with the “querying” processes? If I were using a lower level language, I would ideally just be passing pointers to the existing data around, but that’s not the world of BEAM 
One idea I had was to abuse the fact that Erlang will pass references to large binary blobs, but I don’t think this will be fruitful, and the process of encoding the “split up” result from the bulk query to JSON, sending it, then decoding it, is just a copy operation with extra steps. Writing the data to ETS is also likely not a solution, for similar reasons.
Where Next?
Trending in Questions
Other Trending Topics
Latest on Elixir Forum
Sponsor Spotlight
Bringing learning, work and compliance together since 1987.
Categories:
Sub Categories:
Forums
Popular Tags
- #ecto
- #liveview
- #troubleshooting
- #learning-elixir
- #deployment
- #library
- #erlang
- #testing
- #genserver
- #mix
- #absinthe
- #remote-other
- #otp
- #plug
- #how-to-question
- #macros
- #postgres
- #elixirconf
- #channels
- #exunit
- #discussion
- #code-sync
- #javascript
- #podcasts
- #onsite
- #dialyzer
- #docker
- #authentication
- #umbrella
- #full-time-contract
- #podcasts-by-brainlid
- #ecto-query
- #blog-post
- #elixir-ls
- #ai
- #elixirconf-us
- #phoenix_html
- #iex
- #graphql
- #genstage
- #websockets
- #supervisor
- #advent-of-code
- #distillery
- #processes
- #api
- #forms
- #hex
- #security
- #metaprogramming
Our Sponsors
Build Elixir applications with speed and confidence.
Supporting innovation across the BEAM ecosystem.
We build reliable cloud platforms for business-critical systems.
Catch errors, track performance, monitor hosts and more.
Error tracking, logs, uptime, and Just Enough APM. Built for developers.
Practical resources that improve the lives of professional developers.
Master modern tools, write cleaner code, and tackle real-world challenges.
Producing high quality Elixir screencasts since 2017.
The team behind Membrane, Popcorn, LiveDebugger. Available for hire.
Courses that'll move you from confusion to "Aha, now I get it!"
Showing Posts 1 to 10- Show Best Posts
- Show All (oldest first)
- Show All (newest first)
trisolaran
Have you tried scaling out your ES cluster by adding more nodes and sharding and/or replicating the index across them?
dimitarvp
Try
NimblePoolor:jobs. They invert the flow of “checking out” stuff from a pool which means that your querying processes will work directly with the 1M JSON object and will not copy it. This does not come completely for free and I recommend reading theNimblePoolintro that I linked, it’s excellently explained and demonstrated.:jobsdoes something similar with some more features, though it also offers automatic limiting of requests if you need it (I think you don’t, you seem to have a fixed amount of Elastic and Rabbit workers).ollien
NimblePool seems like a really interesting option, but I’m not clear how that would work out here. It seems to me that this will serialize access to a resource (much like a GenServer), and will provide access to the responses within a callback, thereby avoiding the copying. However, it seems to me that there would be no way to “fan out” the response from NimblePool, right? In other words, if I have 250 workers performing queries, they will be forced to operate on the responses one by one, eliminating concurrency (I wouldn’t be able to move the resource back to the worker process without copying). Am I misunderstanding this?
ollien
It’s something we’ve talked about. We’ve been looking to optimize the code first before scaling our infra.
dimitarvp
Not one by one, no.
You can have 250 workers waiting their turn to use e.g. 20 connections to Elastic. It helps with that.
ollien
Ah, I see. This would avoid the need for the “batching” that I’m doing. I’m not sure how that would work out at first blush but let me give it some thought. Thank you!
dimitarvp
Yeah it’s not completely for free, you have to do a thing or two manually, but the selling point of both options I gave you is basically this: they don’t copy stuff.
Interested to hear about the solution you’ll settle on btw, hope you post it in the future.
ollien
If it ends up being anything less than a rearchitecture, I’d be happy to
JEG2
Instead of copying the data to the processors, could the processors send functions to the data holders?
ollien
Heh, so yes, they can (and do; I actually use a callback mechanism to introduce the
sendcalls). However, dispatching that data to be processed non-serially would require copying, no? It’s kind of a catch-22; either I process them serially and take that performance hit, or I take the performance hit of copying everything and then run my processing concurrently. One could avoid copying by just doing something likeEnum.each(data, callback), but that would be run serially. If I used something likeTask.async_stream/2, the data would still be copied when spawning those procs.