parallelizer

Simplifies creating a pool of workers that execute jobs in parallel

Features

• Easy to use
• Context Support
• Fail fast with errors
• Customizable Pool Size
  • Default number of workers is 10
• Customizable Job Queue Size
  • Default size is 100

Examples

Example 1

Running multiple function calls in parallel without a timeout.

package main
import (
    "fmt"
    "github.com/shomali11/parallelizer"
)
func main() {
    group := parallelizer.NewGroup()
    defer group.Close()
    group.Add(func() error {
        for char := 'a'; char < 'a'+3; char++ {
            fmt.Printf("%c ", char)
        }
        return nil
    })
    group.Add(func() error {
        for number := 1; number < 4; number++ {
            fmt.Printf("%d ", number)
        }
        return nil
    })
    err := group.Wait()
    fmt.Println()
    fmt.Println("Done")
    fmt.Printf("Error: %v", err)
}

Output:

a 1 b 2 c 3 
Done
Error: <nil>

Example 2

Running multiple slow function calls in parallel with a context with a short timeout.
Note: The timeout will not kill the routines. It will just stop waiting for them to finish

package main
import (
    "context"
    "fmt"
    "github.com/shomali11/parallelizer"
    "time"
)
func main() {
    group := parallelizer.NewGroup()
    defer group.Close()
    group.Add(func() error {
        time.Sleep(2 * time.Second)
        fmt.Println("Finished work 1")
        return nil
    })
    group.Add(func() error {
        time.Sleep(2 * time.Second)
        fmt.Println("Finished work 2")
        return nil
    })
    ctx, cancel := context.WithTimeout(context.Background(), time.Second)
    defer cancel()
    err := group.Wait(parallelizer.WithContext(ctx))
    fmt.Println("Done")
    fmt.Printf("Error: %v", err)
    fmt.Println()
    time.Sleep(2 * time.Second)
}

Output:

Done
Error: context deadline exceeded
Finished work 2
Finished work 1

Example 3

Running multiple function calls in parallel with a large enough worker pool.

package main
import (
    "fmt"
    "github.com/shomali11/parallelizer"
)
func main() {
    group := parallelizer.NewGroup(parallelizer.WithPoolSize(10))
    defer group.Close()
    for i := 1; i <= 10; i++ {
        i := i
        group.Add(func() error {
            fmt.Print(i, " ")
            return nil
        })
    }
    err := group.Wait()
    fmt.Println()
    fmt.Println("Done")
    fmt.Printf("Error: %v", err)
}

Output:

7 6 3 2 8 9 5 10 1 4  
Done
Error: <nil>

Example 4

Running multiple function calls with 1 worker. Note: the functions are no longer executed in parallel but sequentially

package main
import (
    "fmt"
    "github.com/shomali11/parallelizer"
)
func main() {
    group := parallelizer.NewGroup(parallelizer.WithPoolSize(1))
    defer group.Close()
    for i := 1; i <= 10; i++ {
        i := i
        group.Add(func() error {
            fmt.Print(i, " ")
            return nil
        })
    }
    err := group.Wait()
    fmt.Println()
    fmt.Println("Done")
    fmt.Printf("Error: %v", err)
}

Output:

1 2 3 4 5 6 7 8 9 10 
Done
Error: <nil>

Example 5

Running multiple function calls in parallel with a small worker pool and job queue size. Note: the Add call blocks until there is space to push into the Job Queue

package main
import (
    "fmt"
    "github.com/shomali11/parallelizer"
    "time"
)
func main() {
    group := parallelizer.NewGroup(parallelizer.WithPoolSize(1), parallelizer.WithJobQueueSize(1))
    defer group.Close()
    for i := 1; i <= 10; i++ {
        group.Add(func() error {
            time.Sleep(time.Second)
            return nil
        })
        fmt.Println("Job added at", time.Now().Format("04:05"))
    }
    err := group.Wait()
    fmt.Println()
    fmt.Println("Done")
    fmt.Printf("Error: %v", err)
}

Output:

Job added at 00:12
Job added at 00:13
Job added at 00:14
Job added at 00:15
Job added at 00:16
Job added at 00:17
Job added at 00:18
Job added at 00:19
Job added at 00:20
Job added at 00:21
Done
Error: <nil>

Example 6

Running multiple function calls in parallel with a large enough worker pool and job queue size. Note: In here the Add calls did not block because there was plenty of space in the Job Queue

package main
import (
    "fmt"
    "github.com/shomali11/parallelizer"
    "time"
)
func main() {
    group := parallelizer.NewGroup(parallelizer.WithPoolSize(10), parallelizer.WithJobQueueSize(10))
    defer group.Close()
    for i := 1; i <= 10; i++ {
        group.Add(func() error {
            time.Sleep(time.Second)
            return nil
        })
        fmt.Println("Job added at", time.Now().Format("04:05"))
    }
    err := group.Wait()
    fmt.Println()
    fmt.Println("Done")
    fmt.Printf("Error: %v", err)
}

Output:

Job added at 00:30
Job added at 00:30
Job added at 00:30
Job added at 00:30
Job added at 00:30
Job added at 00:30
Job added at 00:30
Job added at 00:30
Job added at 00:30
Job added at 00:30
Done
Error: <nil>

Example 7

Showing an example without calling Wait

package main
import (
    "fmt"
    "github.com/shomali11/parallelizer"
    "time"
)
func main() {
    group := parallelizer.NewGroup()
    defer group.Close()
    group.Add(func() error {
        fmt.Println("Finished work")
        return nil
    })
    fmt.Println("We did not wait!")
    time.Sleep(time.Second)
}

Output:

We did not wait!
Finished work

Example 8

Showing an example with a mixture of Add and Wait calls.

package main
import (
    "fmt"
    "github.com/shomali11/parallelizer"
)
func main() {
    group := parallelizer.NewGroup()
    defer group.Close()
    group.Add(func() error {
        fmt.Println("Worker 1")
        return nil
    })
    group.Add(func() error {
        fmt.Println("Worker 2")
        return nil
    })
    fmt.Println("Waiting for workers 1 and 2 to finish")
    group.Wait()
    fmt.Println("Workers 1 and 2 have finished")
    group.Add(func() error {
        fmt.Println("Worker 3")
        return nil
    })
    fmt.Println("Waiting for worker 3 to finish")
    group.Wait()
    fmt.Println("Worker 3 has finished")
}

Output:

Waiting for workers 1 and 2 to finish
Worker 1
Worker 2
Workers 1 and 2 have finished
Waiting for worker 3 to finish
Worker 3
Worker 3 has finished

Example 9

Showing an example with a failed task.

package main
import (
    "errors"
    "fmt"
    "time"
    "github.com/shomali11/parallelizer"
)
func main() {
    group := parallelizer.NewGroup()
    defer group.Close()
    group.Add(func() error {
        return errors.New("something went wrong")
    })
    group.Add(func() error {
        time.Sleep(10 * time.Second)
        return nil
    })
    err := group.Wait()
    fmt.Println()
    fmt.Println("Done")
    fmt.Printf("Error: %v", err)
}

Output:

Done
Error: something went wrong