Resolve Race Condition problems with Map-Reduce and Coroutines

Following up on the Kotlin Coroutines Basic Concepts post, in this article, I will leverage those concepts to address a classic concurrency problem: Race Conditions, using the Map-Reduce pattern. Although concise, this post covers two fascinating topics—Race Conditions and Map-Reduce—so make sure to read through to the end! ;))

Table of Contents:

1. What is a Race Condition?
2. How to resolve Race Conditions?
3. What is Map-Reduce?
4. Resolve Race Conditions with Map-Reduce

1. What is a Race Condition?

Take a look at the following GIF:

Without needing extensive explanations, you can easily see that the shared resource can become inconsistent or corrupted if multiple threads contend for it simultaneously.

2. How to resolve Race Conditions?

The root cause of a Race Condition is contention over a shared resource. Solving contention eliminates the race condition. There are two primary schools of thought:

1. Using Synchronizations^[1] or Atomic Operations^[2]

2. Avoiding shared state

Returning to the dogs and ball analogy, the two solutions translate to:

1. Keep only 1 ball and force each dog to wait for its turn—the dogs won’t be very happy :joy:.
2. Provide multiple balls so each dog gets its own—everyone is thrilled :muscle:.

→ The choice depends on your system requirements and constraints ;))

In the past, when multi-core CPUs were rare and RAM was scarce, #1 was the common choice. Today, with abundant multi-core hardware, I believe #2 is often the far superior and scalable architecture.

However, certain use cases still strictly require #1, such as bank account balance updates, inventory counters, or stock order books.

3. What is Map-Reduce?

By now, you likely recognize that #2 is the preferred way to mitigate Race Conditions in high-throughput modern systems. But how do we implement it? One prominent technique is Map-Reduce. So what is Map-Reduce?

Map-Reduce was originally introduced by Jeffrey Dean and Sanjay Ghemawat at Google in 2004, serving as the foundational concept behind Hadoop. At its core, it embodies the divide-and-conquer strategy through two primary phases:

1. Map: Partition the input data into smaller chunks and process them independently in parallel.
2. Reduce: Aggregate and consolidate the intermediate results to produce the final outcome.

To visualize the workflow:

This is the classic “Word Count” example from Hadoop MapReduce deep diving and tuning. The goal is counting word occurrences from an input stream. The Map stage splits and maps inputs, while the Reduce stage shuffles and reduces them into total counts.

4. Resolve Race Conditions with Map-Reduce

Now for the main practical demonstration. We will implement the classic Race Condition benchmark of incrementing a counter using 3 approaches: synchronous single-thread, Map-Reduce with Coroutines, and multi-threading with AtomicLong.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

fun main(args: Array<String>) = runBlocking {

  val max = 1000_000_000L // the max value which counter needs to increase to

  // increase using one thread
  var counterSync = 0L
  val timeSync = measureNanoTime { counterSync = increaseSync(max) }
  println("increaseSync: counterSync                        [$counterSync] - time [$timeSync]")

  val workers = 5 // number of workers to split into sub increments

  var counterMapReduce = 0L
  val timeMapReduce = measureNanoTime { counterMapReduce = increaseMapReduce(max, workers) }
  println("increaseSync with Map-Reduce: counterMapReduce   [${counterMapReduce}] - time [$timeMapReduce]")

  var atomicCounter = AtomicLong()
  val timeAtomicCounter = measureNanoTime { increaseWithAtomic(max, workers, atomicCounter) }

  println("increaseWithAtomic: atomicCounter                [${atomicCounter.get()}] - time [$timeAtomicCounter]")

}

// simple loop in a single thread
fun increaseSync(max: Long): Long {
  var counter = 0L
  for (i in 0 until (max)) {
    counter++
  }
  return counter
}

// create worker group, increase counter in each worker, then sum results
suspend fun increaseMapReduce(max: Long, workers: Int): Long = runBlocking {
  val maxByWorker = max / workers // max counter for each worker

  val threadPool = newFixedThreadPoolContext(workers, "workerThreadPool") // workers group

  (1..workers)
    .map { async(threadPool) { increaseSync(maxByWorker) } } // create worker jobs
    .map { it.await() } // wait for all worker jobs to finish
    .reduce { acc, current -> acc + current } // sum total counts from each worker
}

// create thread group with shared counter; each thread increments atomic counter and waits for completion
fun increaseWithAtomic(max: Long, workers: Int, atomicCounter: AtomicLong) {
  val maxByWorker = max / workers

  val threads = (1..workers)
    .map { // create thread group
      thread {
        for (i in 0 until maxByWorker) {
          atomicCounter.incrementAndGet() // atomic operation
        }
      }
    }

  for (thread in threads) { // wait for all threads to finish
    thread.join()
  }
}

The output benchmark looks like this:

increaseSync: counterSync                        [1000000000] - time [304001418]
increaseSync with Map-Reduce: counterMapReduce   [1000000000] - time [171316295]
increaseWithAtomic: atomicCounter                [1000000000] - time [19268418252]

Looking at the execution duration, increaseSync with Map-Reduce is the fastest (1x), increaseSync is 1.8x, and surprisingly, increaseWithAtomic is 112.5x slower. While specific numbers vary across machines, this clearly shows that when the computation itself is very fast compared to the overhead of ensuring a shared state remains synchronized, Using Synchronizations or Atomic Operations introduces massive contention penalty.

Conclusion

When addressing Race Conditions today, Avoiding shared state represents the most performant strategy in the vast majority of cases. Map-Reduce is an effective demonstration of this principle, with many other patterns available for complex requirements. Check out upcoming posts in the Kotlin Coroutines series for more concurrent programming challenges and patterns.

The sample code is updated in this repo. Thanks for reading!