Conditionally Run Consecutive Go Routines

Question

I have the following piece of code. I'm trying to run 3 GO routines at the same time never exceeding three. This works as expected, but the code is supposed to be running updates a table in the DB.

So the first routine processes the first 50, then the second 50, and then third 50, and it repeats. I don't want two routines processing the same rows at the same time and due to how long the update takes, this happens almost every time.

To solve this, I started flagging the rows with a new column processing which is a bool. I set it to true for all rows to be updated when the routine starts and sleep the script for 6 seconds to allow the flag to be updated.

This works for a random amount of time, but every now and then, I'll see 2-3 jobs processing the same rows again. I feel like the method I'm using to prevent duplicate updates is a bit janky and was wondering if there was a better way.

stopper := make(chan struct{}, 3)
var counter int
for {
    counter++
    stopper <- struct{}{}
    go func(db *sqlx.DB, c int) {
        fmt.Println("start")
        updateTables(db)
        fmt.Println("stop"b)
        <-stopper
    }(db, counter)
    time.Sleep(6 * time.Second)

}

in updateTables

var ids[]string
err := sqlx.Select(db, &data, `select * from table_data where processing = false `)
    if err != nil {
        panic(err)
    }

    for _, row:= range data{
        list = append(ids, row.Id)
    }
    if len(rows) == 0 {
        return
    }

    for _, row:= range data{
        _, err = db.Exec(`update table_data set processing = true where id = $1, row.Id)
        if err != nil {
            panic(err)
        }
    }
    // Additional row processing

user_6580047 · Answer

I think there's a misunderstanding on approach to go routines in this case.

Go routines to do these kind of work should be approached like worker Threads, using channels as the communication method in between the main routine (which will be doing the synchronization) and the worker go routines (which will be doing the actual job).

package main

import (
    "log"
    "sync"
    "time"
)

type record struct {
    id int
}

func main() {
    const WORKER_COUNT = 10

    recordschan := make(chan record)

    var wg sync.WaitGroup
    for k := 0; k < WORKER_COUNT; k++ {
        wg.Add(1)
        // Create the worker which will be doing the updates
        go func(workerID int) {
            defer wg.Done() // Marking the worker as done
            for record := range recordschan {
                updateRecord(record)
                log.Printf("req %d processed by worker %d", record.id, workerID)
            }
        }(k)
    }

    // Feeding the records channel
    for _, record := range fetchRecords() {
        recordschan <- record
    }

    // Closing our channel as we're not using it anymore
    close(recordschan)

    // Waiting for all the go routines to finish
    wg.Wait()

    log.Println("we're done!")
}

func fetchRecords() []record {
    result := []record{}
    for k := 0; k < 100; k++ {
        result = append(result, record{k})
    }
    return result
}

func updateRecord(req record) {
    time.Sleep(200 * time.Millisecond)
}

You can even buffer things in the main go routine if you need to update all the 50 tables at once.

Conditionally Run Consecutive Go Routines

1 Answers