DEV Community: Roshan Chokshi

Using OpenMPI for Parallel Processing of Text Files

Roshan Chokshi — Sat, 17 Dec 2022 04:18:00 +0000

What is this about?

This article provides a clear overview of OpenMPI and how it can be used to improve the performance of distributed computing tasks. It also provides a detailed code example that demonstrates how to use OpenMPI to calculate the frequency of each letter in a text file. The article also mentions the drawbacks of the code example, which is useful to consider when deciding if OpenMPI is a good solution for a particular problem. Overall, this is a article that provides a good understanding of OpenMPI.

What is OpenMPI?

OpenMPI is an open source implementation of the Message Passing Interface (MPI) standard. It is a library that provides high-performance communication among computers in a distributed system. The OpenMPI library includes a set of routines for point-to-point communication, collective communication, and remote memory operations.

Point-to-point communication allows two processes to communicate directly with each other. This is useful for exchanging data between two processes, such as a client and a server. Collective communication allows multiple processes to communicate at once. This is useful for tasks that require multiple processes to cooperate, such as sorting an array. Remote memory operations provide a way for processes to access data stored in memory on another process. This is useful for tasks that require sharing data between processes, such as a distributed search algorithm.

The main functionality of the OpenMPI library is to provide communication between processes. This communication is done through message passing. In message passing, each process sends a message to another process, which then receives the message and processes it. Message passing is an important part of distributed computing, as it allows processes to exchange data and coordinate their activities. It also allows for synchronization between processes, which is critical for distributed systems.

The OpenMPI library also provides support for other features, such as parallel programming, process management, and resource management. Parallel programming is a way of writing code that allows multiple processes to run concurrently. This can significantly speed up the execution of tasks. Process management allows processes to be created, managed, and monitored. Resource management allows processes to acquire and release resources, such as memory, in a distributed system.

What are we doing?

This is a program that uses MPI (Message Passing Interface) to perform different operations on a given text file in parallel. It first reads the text file, removes punctuations and converts it to lower case, then partitions the text into four groups based on their colors (Blue, Yellow, Green, Red). Each group performs a specific operation on the text:

Blue group removes unwanted words from the text.
Yellow group sets each word to a separate line.
Green group counts the frequency of each word in the text.
Red group sorts the words by their frequency.

The program uses MPI_Intercomm_create function to create an intercommunicator from two intracommunicators. The intercommunicator allows the groups to communicate and pass data to each other. The program then outputs the results to a file named "out.txt".

Implementation

This code is a C++ implementation of a program that partitions a number of processes into four groups (Blue, Yellow, Green, Red) and calculates the frequency of each letter in a text file. The program takes in two arguments, the input filename and the output filename, and uses MPI to split the text file into four parts. After processing each part, the results are combined and written to the output file.

#include <mpi.h>
#include <stdio.h>

#include <iostream>
#include <string>
#include <chrono>

using namespace std;

typedef std::chrono::high_resolution_clock::time_point TimeVar;

#define duration(a) std::chrono::duration_cast<std::chrono::nanoseconds>(a).count()
#define timeNow() std::chrono::high_resolution_clock::now()

The program begins by including the mpi.h header, followed by the iostream and string library for string manipulation, and the chrono library for timing. The removePunctuation() method removes all punctuation from a string, the convertToLower() method converts all uppercase letters to lowercase, the calculateFrequency() method calculates the frequency of each letter in a string, and the create_Intercommunicator() method creates an intercommunicator between two groups of processes.

Methods:
removePunctuation():

/* remove the unwanted character*/
void removePunctuation(char *chunk)
{
    int count = 0;
    int i = 0, j = 0;
    for (i = 0; chunk[i] != '\0'; i++)
    {
        if ((chunk[i] >= 'a' && chunk[i] <= 'z') || (chunk[i] >= 'A' && chunk[i] <= 'Z'))
        {
            chunk[j] = chunk[i];
            j++;
        }
    }

    // add null character at the end of the string
    chunk[j] = '\0';
}

convertToLower():

/* convert the string to lower case*/
void convertToLower(char *chunk)
{
    int i = 0;
    while (chunk[i] != '\0')
    {
        if (chunk[i] >= 'A' && chunk[i] <= 'Z')
        {
            chunk[i] = chunk[i] + 32;
        }
        i++;
    }
}

calculateFrequency():

/* calculate character frequency*/
int calculateFrequency(char *chunk, int *freq)
{
    int i = 0;
    while (chunk[i] != '\0')
    {
        if (chunk[i] >= 'a' && chunk[i] <= 'z')
        {
            freq[chunk[i] - 'a']++;
        }
        i++;
    }

    return *freq;
}

create_Intercommunicator():

void create_Intercommunicator(MPI_Comm &comm, int color, MPI_Comm &a, MPI_Comm &b, MPI_Comm &c, MPI_Comm &d)
{
    if (color == 0)
    { // Blue: remove the unwanted word in text file
        /* Creates an intercommunicator from two intracommunicators. */
        MPI_Intercomm_create(comm, 0, MPI_COMM_WORLD, 1, 1, &a);
    }
    else if (color == 1)
    { // Yellow
        // set to one word each line
        MPI_Intercomm_create(comm, 0, MPI_COMM_WORLD, 0, 1, &a);
        MPI_Intercomm_create(comm, 0, MPI_COMM_WORLD, 2, 12, &b);
    }
    else if (color == 2)
    { // Green
        // count the word frequency
        MPI_Intercomm_create(comm, 0, MPI_COMM_WORLD, 1, 12, &b);
        MPI_Intercomm_create(comm, 0, MPI_COMM_WORLD, 3, 123, &c);
    }
    else if (color == 3)
    { // Red
        // sort by the word frequency
        MPI_Intercomm_create(comm, 0, MPI_COMM_WORLD, 2, 123, &c);
    }
}

The main() method begins by initializing the MPI environment and declaring the colors, communicators, and file variables.

int main(int argc, char const *argv[])
{
    TimeVar t1 = timeNow();
    string colors[4] = {"Blue", "Yellow", "Green", "Red"};
    MPI_Comm group_comm, BY_comm, YG_comm, GR_comm, RD_comm;
    MPI_File in, out;
    MPI_Offset filesize; /*  integer type of size sufficient to represent the size (in bytes) */
    MPI_Status status;   /* Status returned from read */
    int world_rank, world_size, group_rank, group_size;
    int initialized, finalized;
    int ierr;
    int bufsize;
    int freqsum[26] = {0};

    MPI_Initialized(&initialized);
    if (!initialized)
        MPI_Init(NULL, NULL);

    MPI_Comm_rank(MPI_COMM_WORLD, &world_rank);
    MPI_Comm_size(MPI_COMM_WORLD, &world_size);

It then checks the number of arguments and exits with an error message if there are not enough. The code then opens the input and output files and gets their sizes.


    /* Check the arguments */
    if (argc != 3)
    {
        if (world_rank == 0)
            fprintf(stderr, "Usage: %s infilename outfilename\n", argv[0]);
        MPI_Finalize();
        exit(1);
    }

    /* Read the input file */
    ierr = MPI_File_open(MPI_COMM_WORLD, argv[1], MPI_MODE_RDONLY, MPI_INFO_NULL, &in);
    if (ierr)
    {
        if (world_rank == 0)
            fprintf(stderr, "%s: Couldn't open file %s\n", argv[0], argv[1]);
        MPI_Finalize();
        exit(2);
    }

    /* Get the size of the file */
    ierr = MPI_File_get_size(in, &filesize);
    if (ierr)
    {
        if (world_rank == 0)
            fprintf(stderr, "%s: Couldn't read file size of %s\n", argv[0], argv[1]);
        MPI_Finalize();
        exit(3);
    }

    /* Open the output file */
    ierr = MPI_File_open(MPI_COMM_WORLD, argv[2], MPI_MODE_CREATE | MPI_MODE_WRONLY, MPI_INFO_NULL, &out);
    if (ierr)
    {
        if (world_rank == 0)
            fprintf(stderr, "%s: Couldn't open output file %s\n", argv[0], argv[2]);
        MPI_Finalize();
        exit(4);
    }

The code then splits the MPI_COMM_WORLD into four groups based on the rank of the process and creates an intercommunicator between each group. The process then sets the view for the input and output files.


    /* split into groups*/
    int color = world_rank % 4;
    MPI_Comm_split(MPI_COMM_WORLD, color, world_rank, &group_comm);
    MPI_Comm_rank(group_comm, &group_rank);

    MPI_Comm_size(group_comm, &group_size);

    printf("Rank %d/%d in original comm, group [%s] %d/%d in new comm\n", world_rank, world_size, colors[color].c_str(), group_rank, group_size);

    /* Calculate how many elements that is */
    filesize = filesize / sizeof(char);
    /* Calculate how many elements each processor gets */
    bufsize = filesize / group_size;

    ierr = MPI_File_set_view(in, group_rank * bufsize, MPI_CHAR, MPI_CHAR, "native", MPI_INFO_NULL); // split the file for group members
    if (ierr)
    {
        if (group_rank == 0)
            fprintf(stderr, "%s: Couldn't set file view for %s", argv[0], argv[1]);
        MPI_Finalize();
        exit(5);
    }

    ierr = MPI_File_set_view(out, group_rank * bufsize, MPI_CHAR, MPI_CHAR, "native", MPI_INFO_NULL); // split the file for group members
    if (ierr)
    {
        if (group_rank == 0)
            fprintf(stderr, "%s: Couldn't set file view for %s", argv[0], argv[1]);
        MPI_Finalize();
        exit(6);
    }

    /* create intercommunicator for groups */
    create_Intercommunicator(group_comm, color, BY_comm, YG_comm, GR_comm, RD_comm);

    MPI_Barrier(MPI_COMM_WORLD);

Reads the input file, removes punctuation, converts all letters to lowercase, calculates the frequency of each letter, and writes the results to the output file. Finally, the program closes the input and output files and exits with the total time taken.

if (color == 0)
    { // Blue: remove the puctuations and spaces in text file
        int nrchar = 0;
        char *buf = new char[bufsize + 1];
        /* Reads a file starting at the location specified by the individual file pointer (blocking, noncollective) */
        ierr = MPI_File_read(in, buf, bufsize, MPI_CHAR, &status);
        if (ierr)
        {
            if (group_rank == 0)
                fprintf(stderr, "%s: Couldn't read from file %s", argv[0], argv[1]);
            MPI_Finalize();
            exit(7);
        }
        /* Gets the number of top-level elements received. */
        MPI_Get_count(&status, MPI_CHAR, &nrchar);
        /* Add a null character to the end of the buffer */
        buf[nrchar] = '\0';
        removePunctuation(buf);
        MPI_Send(buf, bufsize, MPI_CHAR, group_rank, 0, BY_comm);
    }
    else if (color == 1)
    { // Yellow: convert to lowercase
        char *buf = new char[bufsize + 1];
        /* receive the data from blue */
        MPI_Recv(buf, bufsize, MPI_CHAR, group_rank, 0, BY_comm, &status);
        convertToLower(buf);
        MPI_Send(buf, bufsize, MPI_CHAR, group_rank, 0, YG_comm);
    }
    else if (color == 2)
    { // Green: calculate the char frequency
        char *buf = new char[bufsize + 1];
        int freq[26] = {0};
        /* receive the data from yellow */
        MPI_Recv(buf, bufsize, MPI_CHAR, group_rank, 0, YG_comm, &status);
        calculateFrequency(buf, freq);
        MPI_Send(freq, 26, MPI_INT, group_rank, 0, GR_comm);
    }
    else if (color == 3)
    { // Red: write the result to file
        int freq[26] = {0};
        MPI_Recv(freq, 26, MPI_INT, group_rank, 0, GR_comm, &status);
        MPI_Reduce(freq, freqsum, 26, MPI_INT, MPI_SUM, 0, group_comm);
        if (group_rank == 0)
        {
            char outbuf[1000];
            int index = 0;
            for (int i = 0; i < 26; i++)
            {
                index += sprintf(outbuf + index, "%c: %d\n", 'a' + i, freqsum[i]);
            }

            ierr = MPI_File_write(out, outbuf, index, MPI_CHAR, &status);
            if (ierr)
            {
                if (group_rank == 0)
                    fprintf(stderr, "%s: Couldn't write to file %s", argv[0], argv[2]);
                MPI_Finalize();
                exit(8);
            }
        }
    }

    MPI_File_close(&in);
    MPI_File_close(&out);

    MPI_Finalized(&finalized);
    if (!finalized)
        MPI_Finalize();
    cout << "Total Time: " << duration(timeNow() - t1) << " seconds";
    return 0;
}

How to run?

To run this program, you need to have MPI installed on your system. You can then compile the program using
mpic++ main.cpp -o main.out
and run it using
mpirun --hostfile mpi.config -np 4 ./main.out test.txt out.txt
Where test.txt is the input file and out.txt is the output file. The --hostfile option specifies the host file that contains a list of hosts (computers) to be used in the MPI job, and -np 4 specifies the number of processes to run.

Results:

Drawbacks:
One of the drawbacks of this code is that it is limited in the number of processes it can use. This is due to the fact that the intercommunicator only supports four processes. This means that if more than four processes are available, the code will not be able to utilize them. Additionally, the code is limited in the size of the file that it can read. This is due to the fact that it splits the file into chunks that are equal in size. If the file is too large, the code will not be able to split it into chunks that are equal in size.

What is an APIView?

Roshan Chokshi — Thu, 12 Aug 2021 18:12:19 +0000

The Django REST framework offers a couple of helper classes we can use to create our API endpoints.

APIView
ViewSet

The APIView and The ViewSet both classes are slightly different and offer their own benefits in this post we will be diving in APIView.

The APIView is the most basic type of view we can use to build our API. It enables us to describe the logic which makes our API endpoint.

An APIView allows us to define functions that match the standard HTTP methods:

GET, POST, PUT, PATCH, DELETE

When to use APIViews?

Need full control over logic.
Processing files and rendering a synchronous response.
Calling other APIs and Services.
Accessing local files or data.

Now let's create our first APIView. First we are going to import some classes in our views.py.

from rest_framework.views import APIView
from rest_framework.response import Response

Creating `FirstAPIView` Class.

Underneath the imports let's create a new class called FirstAPIView and inherit from the APIView which we imported from rest_framework.

class FirstAPIView(APIView):
    """ Test API View """

    def get(self, request, format=None):
        """ Returns some random values """
        py_list = [
            "apples","bananas",2,5
            ]

        return Response({'message': 'Hello!', 'list': py_list})

We created a get function which will return message and list as response if we receive GET request on this path.

Wiring it to URl.

from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('admin/', admin.site.urls),
    path('api/', include('APP_NAME.urls')),
]

In above code we have made changes over boilerplate code such as:

Importing include function which helps us to include Urls from other apps.
Created new path api/ and wire it up with urls file of app.

Create new file in app folder as urls.py and paste below lines of code in it.

from django.urls import path
from APP_NAME import views

urlpatterns = [
    path('first-view', views.FirstAPIView.as_view()),
    ]

By adding above code we:

Imported path from django.urls and views from our custom app.
Created urlpatterns which linked first-view path to our FirstAPIView class.

Testing our APIView.

In terminal:

python manage.py runserver 0.0.0.0:8080

Visit below link using any browser:

127.0.0.1:8080/api/first-view/

Demo

How to Use Email as Username for Django Authentication.

Roshan Chokshi — Thu, 05 Aug 2021 13:09:39 +0000

Django comes with a default user model that's used for the standard authentication system. If you had worked on Django you probably have seen this authentication page.

We're going to override this model with our own custom model that allows us to use an Email instead of the standard Username that comes with the Django default model.

We're gonna modify models.py file to include our user profile model.
The first thing we need to do is import some additional classes at the top.

AbstractBaseUser

According to Django documentation AbstractBaseUser has the authentication functionality only , it has no actual fields, you will supply the fields to use when you subclass.

PermissionsMixin

Django provides PermissionsMixin. This is an abstract model you can include in the class hierarchy for your user model, giving you all the methods and database fields necessary to support Django’s permission model.

BaseUserManager

If your user model defines username, email, is_staff, is_active, is_superuser, last_login, and date_joined fields the same as Django’s default user, you can install Django’s UserManager; however, if your user model defines different fields, you’ll need to define a custom manager that extends BaseUserManager.

Add this lines at top of the file.



from django.contrib.auth.models import AbstractBaseUser
from django.contrib.auth.models import PermissionsMixin
from django.contrib.auth.models import BaseUserManager

Creating UserProfile Class.

Underneath the imports let's create a new class called UserProfile and inherit from the AbstractBaseUser and PermissionsMixin



class UserProfile(AbstractBaseUser, PermissionsMixin):
    """ Database model for users in the system """
    email = models.EmailField(max_length=255, unique=True)
    name = models.CharField(max_length=255)

    objects = UserProfileManager()

    USERNAME_FIELD = 'email'
    REQUIRED_FIELDS = ['name']

    def __str__(self):
        """ Return string representation of our user """
        return self.email

By adding above code we:

Made the email and name field required and made email unique
Set the USERNAME_FIELD which defines the unique identifier for the username to email.
Specified that all objects for the class come from the UserProfileManager

Creating UserProfileManager Class.

Now that we have our custom user model we can go ahead and create a manager. Now because we've customized our model we need to tell Django how to interact with this user model in order to create users because by default when it creates a user it expects a user name field and password field but we replace the user name field with an email field.So we just need to create a custom manager that can handle creating users with an email field instead of a user name field.

Let's create a new class called UserProfileManager and inherit from the BaseUserManager.



class UserProfileManager(BaseUserManager):
    """ Manager for user profiles """

UserProfileManager the way it manages work is you specify
some functions within the manager that can be used to manipulate objects within the module that the manager is for.

Below are two functions which we need to add in UserProfileManager class.



def create_user(self, email, name, password=None):
        """ Create a new user profile """
        if not email:
            raise ValueError('User must have an email address')

        email = self.normalize_email(email)
        user = self.model(email=email, name=name)

        user.set_password(password)
        user.save(using=self._db)

        return user

    def create_superuser(self, email, name, password):
        """ Create a new superuser profile """
        user = self.create_user(email,name, password)
        user.is_superuser = True
        user.is_staff = True

        user.save(using=self._db)

        return user

Above code sets logic for when one creates users and superusers using Django CLI.

Update Settings to use Custom Model.

We need add below line in settings.py to use this model over default one.



AUTH_USER_MODEL = 'App_name.UserProfile'

Make changes in Database.

Create new migrations and migrate them which will create a new database that uses our custom User Model:



$ python manage.py makemigrations

$ python manage.py migrate

Create superuser



$ python manage.py createsuperuser

Email address: chokshiroshan@gmail.com

Password:

Password (again):

Superuser created successfully.

Test

As you can see instead of asking for Username it is prompting for Email.