Building a High-Performance Distributed Program with MPI

COMP90025 is Parallel and Multicore Computing at the University of Melbourne. The course has two major assignments that together form a simple parallel spell checker. The first uses OpenMP to collect the unique words in a document. The second switches to MPI, checking those words against a dictionary to find misspellings and suggest corrections.

I wrote this after finishing the second assignment, so the code and results here are from the version I submitted. The code is in xxxbrian/mpi-spell-checker.

The assignment

Running the program is straightforward.

./spellcheck dictionary.txt word_list.txt

dictionary.txt is the sorted dictionary, and word_list.txt contains the words to check. For each word that is not in the dictionary, the program generates every candidate obtainable by replacing, deleting or inserting one character. Any of those candidates found in the dictionary is a possible correction. There is also a capitalization rule: if lowercasing the first letter of a word makes it match a dictionary entry, that word counts as being in the dictionary too.

The output has one line per misspelled word, followed by its candidate corrections, like this.

appel:  appeal apple rappel
stap:  sap snap stab step stop strap tap

The lines are sorted by the number of candidates, from fewest to most, with ties broken lexicographically.

The assignment adds one restriction: each process may retain only part of the dictionary, rather than keeping a complete copy. We were given a single-threaded reference implementation based on a BK-tree, and our program had to run on Spartan, the university's HPC cluster. The grade was split between the code and the report, with the report carrying more weight, so the experiments and analysis mattered just as much.

Change the algorithm first

Before getting started, I ran the reference implementation. It uses a BK-tree to find dictionary words within an edit distance of 1. On the largest dataset, with over 620,000 dictionary entries and over 650,000 words to check, it took about 32 minutes.

My approach was much simpler. Since the edit distance is fixed at 1, I can generate all the candidate words directly and look each one up in a hash table. The candidate generation code is straightforward.

std::vector<std::string> GenerateEditDistance1Candidates(const std::string& word) {
  std::vector<std::string> out;
  const std::string alphabet = "abcdefghijklmnopqrstuvwxyz"
                               "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
                               "0123456789";

  for (size_t i = 0; i < word.size(); ++i)  // 删除
    out.push_back(word.substr(0, i) + word.substr(i + 1));

  for (size_t i = 0; i < word.size(); ++i)  // 替换
    for (char c : alphabet)
      if (c != word[i]) {
        std::string s = word;
        s[i] = c;
        out.push_back(s);
      }

  for (size_t i = 0; i <= word.size(); ++i)  // 插入
    for (char c : alphabet)
      out.push_back(word.substr(0, i) + c + word.substr(i));

  return out;
}

A word of length L produces roughly 124 × L candidates, so an eight- or nine-letter word gives around a thousand. That is a trivial number of hash table lookups. A single process finished in about 20 seconds, with 19 seconds spent generating and checking candidates.

This dataset contains 30627 misspelled words. At over a thousand candidates per word, that adds up to more than 30 million lookups.

So this roughly hundredfold speedup has nothing to do with parallelism. It comes entirely from changing the algorithm. The remaining question is how to divide those 20 seconds among multiple processes.

The first version

The first version was deliberately straightforward. I just wanted to get it running.

Each process read the dictionary itself, then kept its own hash-based partition. To check whether a word was in the dictionary, I called MPI_Allreduce for each word: every process searched its own partition, and the results were combined with a logical OR.

for (size_t i = 0; i < word_list.size(); ++i) {
  bool local_hit = local_dict.Contains(word_list[i]);
  bool global_hit = false;
  MPI_Allreduce(&local_hit, &global_hit, 1, MPI_C_BOOL, MPI_LOR, MPI_COMM_WORLD);
  in_dict[i] = global_hit;
}

Candidate validation followed the same idea. I sent the candidates to every process, had each process look them up, and sent the matches back. Both the sends and receives were blocking.

The basic idea works, but the problems are obvious. A word list with 650,000 entries means 650,000 collective operations. Sending every candidate to every process also multiplies the amount of data. Worse, blocking sends and receives can deadlock: if two processes try to send each other a large block of data at the same time, neither send completes, and the program gets stuck.

I dealt with deadlock first. Initially, I just changed the exchange to MPI_Sendrecv. Later, I replaced the whole validation flow with requests and responses. A process cannot simply wait for a reply, because the other process may be waiting for it too. So each iteration uses MPI_Iprobe to check for incoming requests and handle any that have arrived.

std::vector<int> WaitForResponse(int target, const DistributedDict& dict) {
  while (true) {
    DrainIncomingRequests(dict);  // 等回复的时候，也在帮别人查
    int flag = 0;
    MPI_Iprobe(target, kResponseHeader, MPI_COMM_WORLD, &flag, MPI_STATUS_IGNORE);
    if (flag)
      return ReceiveResponse(target);
  }
}

This keeps a process handling other processes' requests while it waits for a reply, avoiding a stalemate where everyone is waiting and nobody is answering.

File reading came next. Since the assignment requires a distributed dictionary, I switched to MPI-IO so that each process reads only its own section. The file is split by byte offset, though, so a boundary may fall in the middle of a word. I move both the start and end forward to the next newline, ensuring that each word belongs to just one process.

MPI_Offset chunk = file_size / world_size;
MPI_Offset start = rank * chunk;
MPI_Offset end = (rank == world_size - 1) ? file_size : (rank + 1) * chunk;

if (rank > 0)              start = SkipToNextLine(fh, start);
if (rank < world_size - 1) end = SkipToNextLine(fh, end);

MPI_File_read_at(fh, start, buffer, end - start, MPI_CHAR, MPI_STATUS_IGNORE);

This is illustrative code; buffer allocation, EOF handling and large-file counts are omitted. The actual code locates line boundaries by calling MPI_File_read to read forward one byte at a time until it reaches a newline.

Partitioning the dictionary

In the final version, each word has an owner: the process selected by taking the word's hash modulo the number of processes. Dictionary entries are sent to their owners, which store them in their local hash tables.

There is one small detail here. Because of the first-letter capitalization rule, Apple and apple must be handled by the same process. I therefore lowercase the first letter before computing the hash, so either spelling maps to the same process.

int GetOwnerRank(const std::string& word, int world_size) {
  std::string key = word;
  if (!key.empty())
    key[0] = std::tolower(static_cast<unsigned char>(key[0]));  // 首字母统一转小写
  return std::hash<std::string>{}(key) % world_size;
}

The code above only illustrates the ownership rule. The implementation in the report uses FNV-1a for hashing.

The same rule applies to the word list. A word's owner checks whether it is in the dictionary and, if it is misspelled, generates the candidates. Each process has a copy of the word list and takes a round-robin subset of its indices, sending those words to their owners for lookup. The process doing the lookup is also the one that will use the result, so there is no need to broadcast it afterward. As a bonus, the original 650,000 calls to Allreduce have become a single call to Alltoallv.

Checking candidates

Once an owner has its misspelled words, it generates candidates and checks whether they are in the dictionary. A candidate may belong to a different process, though. If it belongs to the local process, a hash table lookup is enough; otherwise, it needs a request to the owner.

The number of potential requests is alarming. With 64 processes, almost all of the more than 36 million candidates belong to another process. Yet the vast majority are not words at all. Sending them out and getting a response is wasted communication.

So I added a Bloom filter for each process. Each process inserts its dictionary partition into a Bloom filter, then uses MPI_Allgather to exchange all the filters, giving every process a copy of everyone else's. Before sending a request, it checks the appropriate filter locally. If the filter says the candidate is absent, it is definitely absent, so there is no need to send it.

for (const std::string& cand : candidates) {
  int owner = GetOwnerRank(cand, world_size);
  if (owner == rank) {
    if (local_dict.Contains(cand)) verified.insert(cand);   // 在本地，直接查
  } else if (MayContain(bloom_of[owner], cand)) {
    remote[owner].push_back(cand);                          // 过滤器说可能在，才发出去
  }                                                         // 说不在的，直接丢掉
}

for (int owner = 0; owner < world_size; ++owner) {
  if (remote[owner].empty()) continue;
  SendCandidateBatch(owner, word_idx, remote[owner]);
  for (int i : WaitForResponse(owner, local_dict))
    verified.insert(remote[owner][i]);
}

Each process's filter has 12 times as many bits as there are words in the entire dictionary: 7.5 million bits, or about 940KB, with 3 hash functions. At 64 processes, each process stores all 64 filters, totaling about 60MB.

The effect was clear: about 99.7% of the more than 36 million candidates were filtered out locally, leaving only 93897 to actually send. I also compared runs with and without the Bloom filters. At 64 processes, the candidate phase took 0.93 seconds without them and 0.31 seconds with them.

Putting it all together, the flow looks like this.

The experiments

The reference implementation comes with a basic test, but that is not enough for measuring performance. I wrote a Python script, batch_srun.py, to run the experiments in batches.

The program records the time spent in each phase, then uses MPI_Reduce to collect the mean, minimum and maximum across processes and print them to standard output. A single-process run looks like this.

[MPI][timing] dict_build avg=0.339734s min=0.339734s max=0.339734s
[MPI][timing] candidate_phase avg=19.1703s min=19.1703s max=19.1703s
[MPI][timing] total_mpi avg=19.9893s min=19.9893s max=19.9893s

The script launches the program with srun, extracts the numbers with a regular expression, and writes them to a CSV file one row at a time.

cmd = ["srun", "-n", str(parallel), "--cpus-per-task=1", "build/spellcheck", dict_path, words_path]
proc = subprocess.run(cmd, text=True, capture_output=True)

RE_TIMING = re.compile(r"^\[MPI\]\[timing\]\s+([a-zA-Z0-9_]+)\s+avg=([0-9.]+)s\s+min=([0-9.]+)s\s+max=([0-9.]+)s")

After every run, the script also compares the output file's MD5 with a known-correct value recorded in advance, and fails immediately if they differ. That means every timing in the table comes from a run with the correct output.

There are five datasets: combinations of dictionaries and word lists containing 10,000 or 100,000 words, plus the largest dataset. I ran each with 1 through 64 processes, for 320 runs in total, all on the same sapphire node on Spartan. The data is partitioned among MPI processes; these experiments do not test scaling across a network of compute nodes.

I only ran each point once, so some noise in the curves is inevitable. The run with 12 processes, for example, was noticeably slower than the neighboring runs with 11 and 13.

Results

Here are the results for the largest dataset, with 620,000 dictionary entries, 650,000 words to check, and Bloom filters enabled. The external script measures the total elapsed time; the internal timing is the program's total_mpi output. The difference includes startup and I/O overhead outside that timed region.

The candidate phase scales very well, dropping from 19 seconds to 0.31 seconds: roughly a 62-fold speedup with 64 processes. The whole program does not scale that far, though. At 64 processes, the total time is 1.88 seconds, just a 10.9-fold speedup over the single-process run.

The bottlenecks are elsewhere. Once the candidate phase becomes very short, the parts that used to seem insignificant start to dominate.

• Every process parses the word list itself. This consistently takes about 0.09 seconds and does not get shorter as more processes are added.
• Building the dictionary means reading the file and then sending the words to their owners. It does get faster from 1 to 8 processes, dropping from 0.34 seconds to 0.09 seconds, but then starts getting slower, reaching 0.51 seconds at 64 processes. Each process reads a smaller section as the process count rises, but the number of messages grows with the square of that count.
• Startup and I/O overhead outside the internal timing also contribute to the roughly 0.8-second gap between the total and internal times.

So at 64 processes, the candidate phase accounts for less than 30% of the internal time, while building the dictionary accounts for nearly half.

That explains the shape of the curve. Beyond about 32 processes, it is essentially flat, with total times varying between 1.3 and 2 seconds. The fastest run was 1.35 seconds with 38 processes, roughly a 15-fold speedup. By 64 processes, it was slightly slower than at 32.

The proportion of forwarded requests that actually find a match also changes with the process count. Each process's filter has a fixed size, so with fewer processes, each partition is larger, the filters are fuller, and false positives are more frequent. With 2 processes, only 19% of forwarded candidates were actually in the dictionary. That rose to 88% with 8 processes, 98% with 16, and 99.96% with 64.

Compared with the original BK-tree, the runtime went from 32 minutes to under two seconds. Roughly a hundredfold improvement came from changing the algorithm, and the remaining ten- to fifteenfold improvement came from parallelism.

Report

I wrote a full report as part of the assignment, covering the experimental setup, results and analysis. You can preview or download the original PDF below.