Passing a csv.NewWriter() to another func in Golang to Write to file Asynchronously

I am making API calls (potentially thousands in a single job) and as they return and complete, I'd like to be able to write them to a single shared file (say CSV for simplicity) instead of waiting for all of them to complete before writing.

How could I share a single csv.Writer() in a way that effectively writes to a single file shared by many threads. This may be too daunting of a task, but I was curious if there was a way to go about it.

package main
import (
    "encoding/csv"
    "os"
)

type Row struct {
    Field1 string
    Field2 string
}

func main () {
    file, _ := os.Create("file.csv")
    w := csv.NewWriter(file)

    // Some operations to create a slice of Row structs that will contain the rows 
    // To write
    var rowsToWrite []Row
    // Now lets iterate over and write to file
    // Ideally, I'd like to do this in a goroutine but not entirely sure about thread safe writes
    for _, r := range rowsToWrite {
        go func(row, writer) {
            err := writeToFile(row, writer)
            if err != nil {
            // Handle error
            }
        }(r, w)
    }

}

func writeToFile(row Row, writer ???) error {

    // Use the shared writer to maintain where I am at in the file so I can append to the CSV
    if err := w.Write(row); err != nil {
        return err
    }
    return nil

}

I would (personally) not have the same file open for writing at two separate points in the code. Depending on how the OS handles buffered writes, etc., you can end up with "interesting" things happening.

Given how you've described your goals, one might do something like (this is off the top of my head and not rigorously tested):

1. Create a channel to queue blocks of text (I assume) to be written - make(chan []byte, depth) - depth could be tuneable based on some tests you'd run, presumably.
2. Have a goroutine open a filehandle for writing on your file, then read from that queueing channel, writing whatever it gets from the channel to that file
3. you could then have n goroutines writing to the queueing channel, and as long as you don't exceed the channel capacity (outrun your ability to write), you might never need to worry about locks.

If you did want to use locks, then you'd need a sync.Mutex shared between the goroutines responsible for enqueueing.

Season to taste, obviously.

Lots of back and forth on this one for me 🙂

I originally thought you could use the Write() method on a csv.Writer in a goroutine, but there are issues when the buffer flushes to disk as the buffer is being written to... not exactly sure.

Anyways, to get back to what you were originally asking for...

Still using the same setup to download Todo objects from <https://jsonplaceholder.typicode.com>, as an example:

type Todo struct {
	UserID    int    `json:"userId"`
	ID        int    `json:"id"`
	Title     string `json:"title"`
	Completed bool   `json:"completed"`
}

// toRecord converts Todo struct to []string, for writing to CSV.
func (t Todo) toRecord() []string {
	userID := strconv.Itoa(t.UserID)
	id := strconv.Itoa(t.ID)
	completed := strconv.FormatBool(t.Completed)

	return []string{userID, id, t.Title, completed}
}

// getTodo gets endpoint and unmarshalls the response JSON into todo.
func getTodo(endpoint string) (todo Todo) {
	resp, err := http.Get(endpoint)
	if err != nil {
		log.Println("error:", err)
	}
	defer resp.Body.Close()
	json.NewDecoder(resp.Body).Decode(&todo)
	return
}

The following:

• Will start one "parent" goroutine to start filling the todos channel:
  • inside that routine, goroutines will be started for each HTTP request and will send the response Todo on todos
  • the parent will wait till all the request routines are done
  • when they're done, the parent will close the todos channel
• Meanwhile, main has moved on and is ranging over todos, picking a Todo off one-at-a-time and writing it to the CSV.
• When the original, "parent" goroutine finally closes todos, the for-loop will break, the writer does a final Flush(), and the program will complete.

func main() {
	todos := make(chan Todo)

	go func() {
		const nAPICalls = 200
		var wg sync.WaitGroup

		wg.Add(nAPICalls)
		for i := 0; i < nAPICalls; i++ {
			s := fmt.Sprintf("https://jsonplaceholder.typicode.com/todos/%d", i+1)

			go func(x string) {
				todos <- getTodo(x)
				wg.Done()
			}(s)
		}
		wg.Wait()

		close(todos)
	}()

	w := csv.NewWriter(os.Stdout)
	w.Write([]string{"UserID", "ID", "Title", "Completed"})
	for todo := range todos {
		w.Write(todo.toRecord())
	}
	w.Flush()
}