Implementing Parallel Computation in Go with a Custom MapReduce Framework

Why Parallel RPC Calls Are Needed

In real‑world business scenarios, assembling a complex object often requires fetching attributes from multiple RPC services such as product, inventory, price, and marketing services. Serial calls cause response time to grow linearly with the number of RPC calls, so parallelization is required for performance.

Limitations of Simple WaitGroup

While a basic waitGroup can handle simple parallelism, it becomes inadequate when data needs validation, transformation, or aggregation, prompting the use of a more structured MapReduce approach.

Scenario 1: Fetching Product Details

func productDetail(uid, pid int64) (*ProductDetail, error) {
    var pd ProductDetail
    err := mr.Finish(
        func() (err error) { pd.User, err = userRpc.User(uid); return },
        func() (err error) { pd.Store, err = storeRpc.Store(pid); return },
        func() (err error) { pd.Order, err = orderRpc.Order(pid); return },
    )
    if err != nil {
        log.Printf("product detail error: %v", err)
        return nil, err
    }
    return &pd, nil
}

Scenario 2: Cleaning a Batch of UIDs

func TestMapReduce(t *testing.T) {
    uids := []int64{1, 2, 3, 4, 5, 6, 79}
    r, err := MapReduce(
        func(source chan<- interface{}) {
            for _, uid := range uids { source <- uid }
        },
        func(item interface{}, writer Writer, cancel func(error)) {
            uid := item.(int64)
            ok, err := check(uid)
            if err != nil { cancel(err) }
            if ok { writer.Write(uid) }
        },
        func(pipe <-chan interface{}, writer Writer, cancel func(error)) {
            var uids []int64
            for p := range pipe { uids = append(uids, p.(int64)) }
            writer.Write(uids)
        },
    )
    assert.Nil(t, err)
    assert.Equal(t, []int64{79}, r)
}

func check(uid int64) (bool, error) {
    if uid > 6 { return true, nil }
    return false, nil
}

MapReduce Code Architecture

generate

– user‑implemented data‑generation function. mapper – user‑implemented data‑processing function. reducer – user‑implemented aggregation function. opts ...Option – configuration, e.g., number of worker goroutines.

Core Function: MapReduce

// MapReduce reads data from source, processes it concurrently with mapper, stores intermediate results in collector, and aggregates them with reducer.
func MapReduce(generate GenerateFunc, mapper MapperFunc, reducer ReducerFunc, opts ...Option) (interface{}, error) {
    source := buildSource(generate)
    return MapReduceWithSource(source, mapper, reducer, opts...)
}

buildSource Function

func buildSource(generate GenerateFunc) chan interface{} {
    source := make(chan interface{})
    threading.GoSafe(func() {
        defer close(source)
        generate(source)
    })
    return source
}

MapReduceWithSource Function

func MapReduceWithSource(source <-chan interface{}, mapper MapperFunc, reducer ReducerFunc, opts ...Option) (interface{}, error) {
    options := buildOptions(opts...)
    output := make(chan interface{})
    defer func() { for range output { panic("more than one element written in reducer") } }()
    collector := make(chan interface{}, options.workers)
    done := syncx.NewDoneChan()
    writer := newGuardedWriter(output, done.Done())
    var closeOnce sync.Once
    var retErr error
    finish := func() { closeOnce.Do(func() { done.Close(); close(output) }) }
    cancel := once(func(err error) {
        if err != nil { retErr = err } else { retErr = ErrCancelWithNil }
        drain(source)
        finish()
    })
    go func() { defer func() { drain(collector); if r := recover(); r != nil { cancel(fmt.Errorf("%v", r)) } else { finish() } }(); reducer(collector, writer, cancel) }()
    go executeMappers(func(item interface{}, w Writer, cancel func(error)) { mapper(item, w, cancel) }, source, collector, done.Done(), options.workers)
    value, ok := <-output
    if err := retErr; err != nil { return nil, err }
    if ok { return value, nil }
    return nil, ErrReduceNoOutput
}

drain Function

// drain blocks until the channel is closed and continuously reads to allow the writer to proceed.
func drain(channel <-chan interface{}) {
    for range channel {}
}

executeMappers Function

func executeMappers(mapper MapFunc, input <-chan interface{}, collector chan<- interface{}, done <-chan PlaceholderType, workers int) {
    var wg sync.WaitGroup
    defer func() { wg.Wait(); close(collector) }()
    pool := make(chan PlaceholderType, workers)
    writer := newGuardedWriter(collector, done)
    for {
        select {
        case <-done:
            return
        case pool <- Placeholder:
            item, ok := <-input
            if !ok { <-pool; return }
            wg.Add(1)
            threading.GoSafe(func() {
                defer func() { wg.Done(); <-pool }()
                mapper(item, writer)
            })
        }
    }
}

The article also includes a diagram illustrating the reducer logic and a test for the drain function that shows how un‑drained channels can cause goroutine leaks.