DEV Community: Volodymyr Potiichuk

Creation of a job processing library using Elixir and Redis

Volodymyr Potiichuk — Sat, 25 Jan 2025 17:28:33 +0000

Today I will cover how you can build your job processing library and understand the core aspects of creating such a tool. This article adds to the previous one about queue data structure. I want to demonstrate a real-world example of using queues and build a job processing library that others can use in their projects.

A tool that we are going to build

Today we are going to build a library that will be able to execute tasks from a queue by a separate process (worker), store these tasks persistently in Redis, and have the ability to check statistics of processed, failed, and scheduled tasks.

To develop this library, we need to familiarize ourselves with the terminology:

Job

Simply put, a job is any function we put in the queue and want to execute.

Worker

A worker is an independent process that performs jobs that are queued. Usually, there can be several workers to process tasks faster, but for simplicity of the code we will create functionality for only one worker, but it will be more than enough.

Prerequisites

There are several prerequisites for building such a tool, and they’re chosen for a reason, let’s look at why:

Redis

Redis is an open-source, in-memory application that we will use as a database for data persistence. The main advantage of Redis is the speed of executing operations because it works straight in RAM (Random Access Memory) and you don’t need to write and read a disk whenever you need some data.

Elixir

Elixir is a dynamic, scalable, functional programming language that is especially good for such applications due to its fault-tolerant behavior.

Also, some Elixir libraries will help us on our way:

Redix

Simple client for Redis, to have the ability to write Redis requests in Elixir language.

Jason

Jason is a library that allows you to encode and parse a JSON structure.

UUID

A library that will allow us to generate unique IDs for our tasks so that we can distinguish them by a unique identifier.

Project structure

To build such a library, let’s understand what basic modules we need for our application to work and I assume that we need to create the following ones:

Enqueuer

This module will be responsible for putting the jobs into the queue.

Dequeuer

This module will be responsible for taking jobs from the queue.

Job

The module is responsible for serialization, deserialization, and data preparation for the queue.

Stats

This module will be able to show us statistics on processed, failed, and scheduled tasks.

QueueWorker

This module is responsible for processing jobs from the queues.

Flow

Before we start implementing this via code, I would like to show how our application is supposed to work with two flow diagrams:

Here is how our library will work when the client creates a new task:

The client provides raw data about the task to the Enqueuer module.
The Enqueuer module asks the Job module to construct job data and serialize it to JSON.
The Enqueuer module passes the generated JSON to the Queue module.
The Queue module saves JSON in the database using the Database module.
We send an asynchronous message to the worker that we have a task in this queue, and if the worker is up and available, he can process it.
Return the created job to the client.

And the second one from the queue worker perspective:

Queue worker checks all queues under the library namespace in Redis.
Queue worker takes the first queue he finds and peeks elements from the start of the queue (FIFO principle).
Queue worker parses the job from JSON to struct via the Job module
Queue worker starts to process the parsed job
When the job is completed, the queue worker moves the job to a different queue based on the result, either failed or processed.
Recursively repeat from the second point until the queue is empty, after that we can go to the next queue.
After all tasks are completed, the worker process remains in standby mode, and as soon as a new task appears, it will be ready to process it. If the worker is not enabled at the moment of task enqueue, all such tasks will be processed at the moment of worker startup.

Project initialization

Let’s start with the creation of the project. We will use the Mix tool, which shipped with Elixir, and use the “mix new ” command for project initiation:

mix new processing_library

After command execution, we should get a ready-made set for developing our projects, namely:

folders structure
dependency management
building tooling
test environment

and in my case, there is such structure with such file naming:

After successful project initiation, we can add the dependencies to the “mix.exs” file:

and install them as well with the Mix tool:

mix deps.get

Also, we need to start Redis, I usually do it through the docker engine and docker-compose utility. Still, you can do it through any convenient installer like homebrew or asdf. Anyway, here’s my docker-compose.yml file:

This configuration will start Redis on Linux Alpine inside the container on port 6379 and we will redirect it to our local host on port 6379. To run it, you need to start the container via docker-compose CLI:

docker-compose up redis

And if everything goes well, you will face something like this:

Implementation

I suggest we start our journey with the module responsible for the tasks:

There, we define the job structure and functions accountable for serialization, deserialization, and job creation from raw data. The structure of the job in our case will have the following fields:

params

These are variables that will be passed to the job function as parameters:

worker_module

This is an abstract module responsible for executing a particular task, and every module that inherits it through behavior directive must implement the “perform” function that the queue worker will eventually call:

jid (Job ID)

For each task, we will assign a job ID to be able to distinguish them by a unique identifier.

error

Since a job can also break during execution, saving the error in the job object would be a good idea.

start_at, finish_at

These time stamps will indicate how long it took the worker to complete the task.

Next, we want to create a module that is a simple command wrapper around the Redix library. This library does not provide high-level functions, only Redix.command/3, an alternative to writing commands directly in the redis-cli client. For a better understanding of what we want to do, let’s show an example of viewing an item in a Redis queue:

Instead, we will have a function in the Redis module that will be responsible for peeking elements:

This approach will help us to divide the area of responsibility between files and increase the cohesion of the modules. Good, now let’s take a look at the elements that make up our Redis wrapper module:

Here we use GenServer behavior to maintain a single Redis connection and to be able to manage this process through the Elixir supervisor. For the client to customize where the library should look for the Redis process, we provide the ability to configure this data via config settings:

In the case of our library, we have several variables and the most interesting one is “redis_namespace”, it allows us to distinguish between library redis data and others. For this purpose, we will have special functions that wrap and unwrap Redis keys in the provided library namespace:

Using these methods we will create a set of core functions, which we will use in the future (P.S. There is only the public API for simplicity, but if you want to see the full file, click here):

Next, with our Redis module, we can create an abstraction module for the database:

We are simply delegating our method calls from the database to the Redis module, but why? Why not just call methods from the Redis module directly?

Explanation: Imagine you’re writing a program and you’re relying on the interface Redis provides. At some moment you realize that you need to switch from Redis to PostgreSQL. You start looking at the code and realize that every part of your program depends not on an abstract database, but on Redis, which makes the move very difficult. That’s why we want to hide the implementation details behind a certain interface so that in case of anything, we would have to change only one module instead of the whole program.

The same goes for the queue module:

The output is a simple and clear interface that can be interacted with.

Having an abstraction over the database and the queue, we can start working with the modules that work most closely with the queue, namely the enqueuer and dequeuer modules:

In this module:

We create a job from incoming parameters
Add it to the Redis queue
If the queue to which we add is not a queue reserved by our library (like processed or failed queues), we send the asynchronous message to the Queue Worker and if it is available it will process it.

Something similar is happening with the dequeuer module:

The remove/1 function is responsible for pulling an item from the queue without paying attention to the order using the find_job/1 function, it is very convenient if we want to remove a task from the queue that was already scheduled some time ago. The last dequeue/1 function is passed directly to the Queue module since we are completely satisfied with it.

Now we going to the fun part, we’ll move on to the queue worker module:

At the very beginning, we request all queues available under our library’s namespace and send them for processing:

After that, going through each queue, we synchronously look at the first available task (by FIFO principle), and send it for processing:

At the moment of job processing:

We save timestamps and log information about the job start to the I/O.
Execute job through Kernel.apply/3 function
Save the execution result in one of the queues (either failed or processed).
Start executing the next task in the queue recursively.

That’s all the queue worker does, but it also accepts other signals from outside (e.g. from the Enqueuer module), and can perform such tasks asynchronously thanks to GenServer.cast/2:

Great, now let’s connect our queue worker so it starts when the client application is started. It’s important to clarify that we will disable the worker and database in the test environment so it can be done manually before running the tests:

After adding our processes to the supervisor, we can start our library and start using it. But wait, something is missing… oh… statistics module. Let’s implement it very fast:

Inside this module we check the queues, pick the number of jobs inside them, and provide a map as a result:

%{processed: 105, scheduled: 70, failed: 5}

Finally, the cherry on top will be a public interface for users, which will provide everything users need and remove the necessity to go deep into modules using our library:

Demo

I made a dashboard using Phoenix LiveView and WebSockets to demonstrate how our library works visually. Also, for this purpose, I added a dummy worker module to our project, which will randomly set a delay for our task between 1 and 3 seconds and randomly throw an error as well so that we could see such cases in UI:

in the demo itself, I’ll enqueue 10 real-life tasks with our dummy worker, and let’s see how the queue worker handles them:

This demonstration shows how the queue worker pulls up one scheduled task each time, synchronously executes it, moves it to one of the queues depending on the result, and proceeds to the next one.

Conclusion

At the end of the story, I want to say that it was super exciting to write such a library, I learned and understood a lot of things related to the concept of job processing tools. I’ll leave a link to the repository here in case anyone wants to check out the whole codebase. Thanks for reading, I hope you have learned something new too!

How to use queue data structure in programming

Volodymyr Potiichuk — Fri, 03 Jan 2025 10:54:45 +0000

A queue in classic programming is a data structure (DS) used for storing and managing data in a specific order. In the real-world simulation, you can imagine a queue of people waiting for their order, and people who ordered it first will receive it first and vice-versa.

This concept is used widely for different reasons in programming and the most popular are those:

Task scheduling

In distributed computing, queues contain a list of tasks to be executed by a process (or several processes, also called workers). When a task is created, it is added to the end of the queue, and the task waits for a free worker to pick it up. Queues in task scheduling can be different, and they don’t always follow the FIFO principle. It can also be:

Priority queues (Each task has its priority, typically a plain number)
Delayed queues (Each task has its time of execution, for example, scheduled email letters)
Dead letter queues (Tasks that worker can’t process, remain for further investigation)

Real-time event processing (Message brokers)

Queues in event-driven architecture are used as a core component. It allows asynchronous communication between the components of the systems (producer and consumers). The producer adds a message to the broker’s queue, and then as soon as a consumer is available, it will pull messages out.

Any other system

That requires the creation of a real queue simulation, like a call center or online game lobby.

Interface

At the previous stage, we figured out where the queue is being used, and it would be cool to understand what contract this data structure has.

There are several common methods of this DS:

enqueue

This core method is responsible for adding the item to the queue.

dequeue

This method is responsible for pulling the element from the queue according to the principle that the queue follows.

peek (front or rear)

This method allows us to view the item from the rear or front without pulling it out. This can be useful if we accidentally pull an item and our process that handles it falls, in which case a pulled item may be lost. This method solves that problem since it obliges us to confirm that we have already processed the desired task.

is_empty

This method allows you to check if the queue is empty and it’s useful for handling edge cases like when there are no items to pull.

Implementation

Knowing this information, we can start writing our implementation of this data structure. The easiest way to implement this will be through another data structure, an array. To implement this, I will use Elixir, a dynamic, functional programming language that has absorbed the best programming patterns, and I like it a lot.

Elixir has a great module called GenServer, it allows you to use isolated processes with unified behavior and also provides the ability to store state and execute code asynchronously. Let’s look at all this with a code snippet:

In this screenshot, you can see the client-side implementation of GenServer module, which is responsible for the public interface of our queue which we mentioned before. Let’s go through the code line-by-line:

Line 2 — We use the GenServer behavior which allows us to create an independent process in the Elixir and store the state. init_arg in this case will be passed from the GenServer.start_link/3 function, and it will be an array from the client side or an empty array by default.

Line 8— This function is responsible for starting a separate process, which will hold our queue with a process identifier.

Line 14— Here you can see the presence of all the necessary methods that a public queue interface should have (queue, dequeue, peek, is_empty). These functions will call methods on the server and return the updated result from them.

On the server side, we have all the dirty work, namely how the queue works:

Line 43 — Here you can see how the client call of the enqueue method calls this callback, which in turn operates by adding the necessary element (the usual push to the end of the array).

Line 48 — The dequeue callback checks if the queue is empty, because if it is, the requested item cannot be pulled. If the queue is empty, we return the tuple with an error and a message about the empty queue {:error, :empty}.

Line 58 and 67 — The same as with dequeue, we pull an element by index and make an additional check for array emptiness, so that there would be no runtime error when pulling elements.

Line 72 — Server callback which is responsible for checking the queue for emptiness, the callback calls a private method is_empty that is not available in the public interface.

Result

We’ve implemented our queue using Elixir, which follows the FIFO principle. Let’s take a look at how it works:

Conclusion

Today we learned what a queue is as a data structure in programming, where it is most often used, and how to implement its basic array-based version using Elixir GenServer. In the next article, we will create a library that can enqueue tasks into queues, persist them in the database with Redis, and process those tasks with workers. Thanks for reading and see you soon!

Learn javascript promises. Part 1 — What is a promise?

Volodymyr Potiichuk — Thu, 02 Jan 2025 18:11:09 +0000

Learn javascript promises. Part 1 — What is a promise?

Hey, javascript fans, today I will tell you about promises and how you can learn or understand them better. Many people think that this is a complicated topic and get confused about it, although there is nothing tricky about it, and if you practice it often, you’ll become an expert, I guarantee it. My task today is to help you understand the basics, and to demonstrate the essence of promises and why we need them.

What is a promise and why do you need it in javascript?

Imagine you have pledged to your friends that you will continuously exercise each morning and notify them if you do that. Your promise is not blocking you or your friends from doing their life and they are just waiting for your notification about the pledge’s success or failure to make their conclusions or actions about you.

This small example illustrates the nature of the promise as an event and the two actors that are connected by this event: in this case, it’s the consumer (your friends), and the producer (you):

And if we simplify it a bit, we can say that in javascript we see the same behavior with the Promise class, which gives us the ability to perform asynchronous tasks without blocking the main thread of the javascript and notify subscribers when we finish our promise.

Let’s take an example to figure out how to work with it:

Let’s review this code in order:

We create a promise instance with an executor function (it’s a function that we pass to the Promise class constructor).

The executor function is the basis of the promise, and this function is called instantly when the promise is created.

The executor function takes two positional parameters, the first is the resolve method and the second is the reject method. Both of these methods are responsible for finishing promise in some time.

For example, we promised to get up every morning and exercise for a week. In this case, our promise lasts for a whole week, and if you can’t keep it on day 3, you can call the reject method and everyone who is subscribed to your promise will know about it. Or vice versa, if you keep your promise, then on day 7 you will call the resolve method, which will inform your friends that you have successfully kept your pledge. So you can see that the consumers (your friends) are completely independent of when you tell them the results, they just need to be sure that when they are, they will be able to get them.

The picture below shows the possible promise states, they are “pending”, “fulfilled” and “rejected”. When a promise is just initiated, its status is “pending”. After it is completed, it is either “fulfilled” or “rejected”, depending on which method was called to complete it.

It is also worth noting that a promise executor cannot be completed more than once, all subsequent calls to resolve or reject methods will be ignored.

At the moment when you say you finished your promise (by calling the resolve or reject) method, all those who have subscribed to the promise will receive a notification with payload, which can be retrieved inside two special methods of the promise instance. Those methods are: Promise.prototype.then and Promise.prototype.catch

Using the Promise.prototype.then method, you can handle both positive and negative scenarios at once, as the function accepts two callbacks as in the screenshot above.

With Promise.prototype.catch you can handle only negative scenarios like this:

and leave the Promise.prototype.then for positive scenarios.

Also, promises provide a method that you can use to execute code that should be executed regardless of the result. This method is Promise.prototype.finally:

This method takes nothing and returns nothing, as it is designed to complete or clean up previous operations. For example, to show the message to the user or to hide the loader.

Great, at this point, we’ve covered the basics of promises, so I suggest we move on to a real-world example, like requesting data from the API, transforming it from raw HTTP response to JSON, and outputting it to the console:

As you can see, the result of using it in the real world is not much different from the skills we practiced at the beginning of the article. But still, you may have noticed a couple of new things:

The result from each previous Promise.prototype.then and Promise.prototype.catch handlers are passed to the next Promise handler. It’s good to mention that the Response.json method returns a promise, which in turn, when executed, will return a JSON, and will call the following handler passing that JSON to it.
In the case of Promise.prototype.finally, it is an “invisible” handler, and you can say that it passes the previous response through itself into the next one.

Awesome, I’m sure at this point you have a better understanding of what a promise is, how to use it, and what states it can have. Also, I want to list below the key points that we went over today (cheatsheet):

Summary:

Javascript promise — it is a special object that allows you to execute asynchronous operations and notify consumers of it when it’s done
Promise instance can have 3 states —”pending”, “fulfilled” and “rejected”
There are three main methods of the promise instance — “then”, “catch” and “finally”.
Promise.prototype.then — designed to handle both rejections and successful results
Promise.prototype.catch — designed to handle rejected results
Promise.prototype.finally — designed to make some cleanup work
You can chain your handlers in any order that you want, they will be executed sequentially
The result from the previous promise goes into the next, so to make the promise chain extensible, you must remember to return a promise from handlers every time except for the “finally” handler.

In future parts of this topic, we’ll cover more advanced promise techniques and delve deeper into its capabilities, so subscribe, like, and stay tuned!

How to automate the launch of your terminal processes (fzf + tmux + teamocil)

Volodymyr Potiichuk — Wed, 27 Nov 2024 20:00:17 +0000

Do you know this feeling, when you get so tired after the workday, that you want nothing more than to go to sleep as soon as possible? Or you know how it feels when you do boring things from day to day and they affect you negatively because they don’t do you any good?

If yes, don’t feel lonely, I had the same problem with terminal work until recently, and I ran into it at my job.

Long story short

I’ve been working as a front-end developer for a consulting company for the past couple of years, and just like on any other regular job, I’ve spent 90% of my time working on 1–2 projects.

But, as time went on I got assigned several projects to do (don’t think that my company is the devil and try to get the maximum from my developer soul), but there was a situation where I didn’t have enough workload on any of the projects I worked on.

Problem description

The main point is that each project it’s a set of different processes (e.g. backend, frontend, database, etc) that should be started to get your project up and running. And if you have one project, running a couple of processes is not a problem, because it is done one or at most a couple of times a day.

Now, let’s imagine the situation when I have 3 projects. The required processes are listed in the image below:

As you can see, there are a bunch of processes in each project, and the ports they occupy on my machine overlap a lot (which means I couldn’t run them in parallel even if I wanted to). So every time to switch between projects and run them, I had to close the old one and retype a set of long process commands like this:

cd backend && pipenv --python 3.14.0 && pipenv run gunicorn main:app --workers 4 --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:4000
# ...
cd frontend && npm run dev
# ...
docker-compose up db redis
# ... and so on

to just get my next project up and running. Of course, I was exhausted doing this by typing and searching for the commands every day for each of those projects several times per day…

As you understand, this tedious, useless, and often repeated process will not affect us favorably but will bring unpleasant moments in our work. The time I spent on this switching between projects could be spent sleeping, eating, or resting my eyes, which I think is much cooler. And worth saying that I am a person who believes that everything we do should bring more pleasure, and the work I spend most of my time on is certainly no exception.

At some point in time, I had an idea of how to fix this, as I realized that the list of commands is almost always the same, it runs from the same folders, and it can quietly be automated.

Solution

To solve this problem we need several tools that are good on their own, but if we combine them, we can optimize our workflows by more than 99%. Let’s talk about each of them in more detail and write some code!

Step 1 — Start using tmux application to make your terminal life easier

What is tmux?

In simple words, tmux is an application that can manipulate any number of terminals (processes) in a single window and provides an API for working with them. Take a look at the picture below:

Here you can see that I have a “test” session, with 3 windows opened and in any of them I can open any amount of processes. It makes switching between projects and processes a lot faster than replicating terminal instances just to achieve the same behavior.

Step 2 — Tool for managing tmux sessions easily

The next step will be a tool that can help us achieve fast bootstrapping of the projects (sessions), it can be either teamocil or tmuxinator, these are the most popular ones and in this tutorial, I will use teamocil. Those tools allow you to write a configuration for a specific session and run this session with many processes and windows in one click.

In the screenshot above you can see the structure of the simple configuration file, it has a *.yaml format, and some hierarchy that simply replicates the architecture of tmux.

Such config leads to the ability to use one command for bootstrapping the tmux session:

teamocil <config_name>

and voilà, we will get up and running session with one command:

As you can see, we have already eliminated one of the problems, we don’t need to type and remember all commands, open new terminal windows, and switch folders every time, those magic tools will do it for us.

Step 3 — Fuzzy finder

FZF (Fuzzy finder) is a command-line application that allows you to turn any list into an interactive menu with support for awesome fuzzy search, which you can read more about here.

I think that you should get familiar with this application as soon as possible because it can come in handy everywhere where you need to search for something in a list, think about GitHub branches for example.

Step 4 — Automate with a shell scripting

At this point, we already know how to open sessions in one click and do a smart (fuzzy) search, now we need to put it all together with shell (sh) scripting.

I am sure you are already familiar with this and used some commands such as “change directory” or “copy” and so on. Thanks to shell scripting it is possible to automate many processes, for example, sending messages to Slack, or pulling the database from the environment, and in our case, we can make a script that will allow us to control our sessions via simple GUI and shortcuts without typing any commands.

In my case, I use the bash shell (Bourne-Again SHell) as it is a slightly improved version of sh. But you can also use the “sh” by itself or tools that allow you to translate your code from any desired language into the shell, for example, “zx” from Google, which allows you to write your scripts in node.js and run them as shell scripts.

Coding part

To meet our expectations with the optimization we need to create an application that can do three things:

Start selected session
Delete selected session
Navigate between running sessions

Let’s start with the first point which we can achieve with teamocil and fuzzy finder, FZF will be as a GUI and teamocil as a tool that we will call when a certain session is selected (to bootstrap our session). A simple version of the script will look like this:

# start selected session
teamocil $(ls $HOME/.teamocil/ | awk -F. '{print $1}' | fzf)

Don’t be afraid if you don’t understand it right now, let’s see what we did there:

Here we take the folder with our session configs and get a list of files from there

# Imagine we have such file structure in our $HOME/.teamocil folder:
# - first_project.yml
# - second_project.yml
# - third_project.yml
# - ...*.yml

ls $HOME/.teamocil/

We process these files via the “awk” tool to get only the session name (without a file extension) and pass them to FZF.

awk -F. '{print $1}' | fzf

# In runtime it will look like this
# first_project.yml     first_project
# second_project.yml -> second_project | fzf
# third_project.yml.    third_project

The result we get from the FZF is forwarded to teamocil, which bootstraps the actual session.

Already with this small command, we can start our sessions in an instant like that:

Alright, we finished with the creation of the sessions, and now we can proceed with switching between sessions and deleting them:

# change active session
tmux switch -t $(tmux list-sessions | awk -F: '{print $1}' | fzf)

# delete session
tmux kill-session -t $(tmux list-sessions | awk -F: '{print $1}' | fzf)

You see, it’s almost the same as the previous one, however, we are checking the list of the active sessions with the tmux command and passing it to the “switch” or “kill-session” native command of the tmux depending on the desired behavior. That’s it, demonstration of our script is there:

At this point, we have one last thing to do: unify our scripts into one whole like a multi-tool, where we can create, delete, and switch between sessions. A cool way to do this is to use the tmux “display-popup” command, which allows you to display a window on top of all the processes in tmux, and that’s where I’d like to put the script implementation.

In my tmux.conf:

bind f display-popup -w 75% -h 50% -E "~/.config/fzf_tmux_sessions/bin/fzf_tmux_sessions.sh"

This command will open a tmux popup with a width of 75% and height of 50%, invoke the code from the specified file.

That’s it, with small improvements to our code above we will get something like this:

It turns out that in this case, you can switch between different projects in less than 5 seconds instead of 5-10 minutes. Isn’t that cool, imagine how much time you save if you switch between projects 5 times a day or just bootstrapping large projects every day? If you want to see more, you can find the full code of this script in my GitHub repo.

In conclusion, I can say that this project has taught me quite a lot of skills, and I think that if you want to optimize something in your life, do not hesitate, it is very exciting and educational.

Thank you so much for reading this article, I hope it was helpful to you, see you next time!