Consider the following piece of code:
import (
"sync"
)
const ( // any natural number works for those
WORKERS = 40
BUFSIZE = 10
CACHEAMT = 5
)
var results []int // the storage of data aquired after completing jobs
type job struct {
// something to do
}
func main() {
jobchan := make(chan job, BUFSIZE)
go initAndSendJobs(jobchan)
reschan := startWorkers(jobchan)
processResults(reschan)
}
func initAndSendJobs(jobchan chan job) {
for /* ... */ {
// make and send jobs to workers through channel
}
close(jobchan) //close channel for workers to stop
}
func startWorkers(jobchan chan job) (chan map[int]int) {
reschan := make(chan map[int]int, BUFSIZE)
ressync := &sync.WaitGroup{}
ressync.Add(WORKERS)
for i := 0; i < WORKERS; i++ {
go doJobs(ressync, jobchan, reschan)
}
go func() { // call goroutine to not block execution of startWorkers()
ressync.Wait()
for len(reschan) != 0 {
}
close(reschan)
}()
return reschan
}
func doJobs(ressync *sync.WaitGroup, jobchan chan job, reschan chan map[int]int) {
defer ressync.Done()
for j := range jobchan {
res := &map[int]int
// do some calculations
reschan << *res
}
}
func processResults(reschan chan map[int]int) {
// implemented cache here to speed up result write to results slice
cachesync := &sync.WaitGroup{}
cachesync.Add(CACHEAMT)
cacheDrain := make(chan map[int]int, BUFSIZE)
go func() {
cachesync.Wait()
close(cacheDrain)
}()
for i := 0; i < CACHEAMT; i++ {
go saveToCache(reschan, cacheDrain, cachesync)
}
for res := range reschan {
for key, value := range res {
// don't need mutex here, since this is the only place where I write data
results[key] += value
}
}
// will get here only once the calculations finished and we need to empty caches
for cachedata := range cacheDrain {
for key, value := range cachedata {
results[key] += value
}
}
}
func saveToCache(reschan, cacheDrain chan map[int]int, cachesync *sync.WaitGroup) {
defer cachesync.Done()
cache := make(map[int]int)
for res := range reschan { // exits once ther is no more jobs to do
for key, value := range res {
cache[key] += value
}
}
cacheDrain <- cache
}
Here I make a fixed amount of goroutines (workers) and send them jobs through jobchan channel. After doing specific job they send results to reschan, from which the results are later saved to results slice. The thing is, that result save in the results slice is the bottleneck of the program, which is what I'm trying to solve.
To speed up this process, I set up multiple goroutines and in each of them there is cache which accumulates data from reschan and then returns it through cacheDrain once either calculations are finished or the process was interrupted (implemented in the original code).
The question is - is there anything I can do to:
- Shorten time required to write data to the
resultsslice? - Shorten time required to fully stop? (In the original code if I interrupt the process it will wait for all workers to finish working on their jobs and will save all the results they aquired which might take a long time)