Add concurrency

concurrency/cmd/csp/csv.go

@@ -0,0 +1,41 @@
+package main
+
+import (
+	"encoding/csv"
+	"io"
+	"os"
+	"sync"
+	"time"
+)
+
+// The free tier of the OpenWeatherMap API allows for 60 requests per minute.
+// ∴ we limit our requests to 1 per second.
+var interval = 1 * time.Second
+
+func processCSV(
+	file *os.File, ztchan chan *ZipTemp,
+) {
+	var (
+		limiter = time.Tick(interval)
+		wg      sync.WaitGroup
+	)
+
+	reader := csv.NewReader(file)
+	for {
+		record, err := reader.Read()
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			panic(err)
+		}
+
+		<-limiter
+
+		wg.Add(1)
+		go getZipTemp(&wg, record, ztchan)
+	}
+
+	wg.Wait()
+	close(ztchan)
+}

concurrency/cmd/csp/display.go

@@ -0,0 +1,47 @@
+package main
+
+import (
+	"fmt"
+	"time"
+)
+
+const (
+	title = "TOP TEN ZIP CODES BY HIGHEST CURRENT TEMPERATURE"
+)
+
+func display(ztchan chan []*ZipTemp, done chan struct{}) {
+	for {
+		locations, more := <-ztchan
+		if !more {
+			done <- struct{}{}
+			return
+		}
+
+		// Clear the screen.
+		print("\033[H\033[2J")
+
+		// Print the title and top ten.
+		fmt.Println(title)
+		fmt.Printf(
+			"Last update: %s.\n\n",
+			time.Now().Format(time.RFC1123),
+		)
+
+		for i, zt := range locations {
+			if zt == nil {
+				fmt.Printf("%d. No data\n", i+1)
+
+				continue
+			}
+
+			fmt.Printf(
+				"%d. %.02f°F %s (%s, %s)\n",
+				i+1,
+				zt.Temperature,
+				zt.ZipCode,
+				zt.City,
+				zt.State,
+			)
+		}
+	}
+}

concurrency/cmd/csp/main.go

@@ -0,0 +1,42 @@
+package main
+
+import (
+	"log"
+	"os"
+	"time"
+)
+
+type ZipTemp struct {
+	City        string
+	Latitude    string
+	Longitude   string
+	State       string
+	Temperature float64
+	Time        time.Time
+	ZipCode     string
+}
+
+func main() {
+	file, err := os.Open("./zipcodes-50.csv")
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer file.Close()
+
+	// Set up channels.
+	var (
+		done   = make(chan struct{})
+		ztchan = make(chan *ZipTemp)
+		ttchan = make(chan []*ZipTemp)
+	)
+
+	// Start the rank and display goroutines.
+	go rank(ztchan, ttchan)
+	go display(ttchan, done)
+
+	// Process the CSV.
+	processCSV(file, ztchan)
+
+	// Wait for the last display output.
+	<-done
+}

concurrency/cmd/csp/rank.go

@@ -0,0 +1,50 @@
+package main
+
+func rank(ztchan chan *ZipTemp, ttchan chan []*ZipTemp) {
+	var topTen = make([]*ZipTemp, 10)
+
+	for {
+		zt, more := <-ztchan
+		if !more {
+			close(ttchan)
+			return
+		}
+
+		var newItem bool
+		for i, tt := range topTen {
+			if tt != nil && tt.Temperature > zt.Temperature {
+				continue
+			}
+
+			// We've found a zip code with a higher temperature.
+			// Insert it into the top ten.
+			topTen = append(topTen, nil)
+			copy(topTen[i+1:], topTen[i:])
+			topTen[i] = zt
+
+			// Drop the last value.
+			topTen = topTen[:10]
+
+			newItem = true
+			break
+		}
+
+		// If the top ten hasn't changed, process the next zip code.
+		if !newItem {
+			continue
+		}
+
+		// If the top ten has changed, message the display goroutine.
+		// Send a copy, to prevent race conditions.
+		var newTopTen = make([]*ZipTemp, 10)
+
+		for i, tt := range topTen {
+			if tt == nil {
+				continue
+			}
+			newTopTen[i] = tt
+		}
+
+		ttchan <- newTopTen
+	}
+}

concurrency/cmd/csp/run.sh

@@ -0,0 +1,5 @@
+#!/bin/bash
+
+export OPEN_WEATHER_MAP_API_KEY="$(pass open-weather-map/api-key)"
+
+go run ./*.go

concurrency/cmd/csp/weather.go

@@ -0,0 +1,38 @@
+package main
+
+import (
+	"sync"
+	"time"
+
+	"github.com/dwrz/talks/concurrency/pkg/weather"
+)
+
+func getZipTemp(
+	wg *sync.WaitGroup, record []string, ztchan chan *ZipTemp,
+) {
+	defer wg.Done()
+
+	// Ignore malformed records.
+	if len(record) < 5 {
+		return
+	}
+
+	var zt = &ZipTemp{
+		City:      record[3],
+		Latitude:  record[1],
+		Longitude: record[2],
+		State:     record[4],
+		ZipCode:   record[0],
+	}
+
+	// Ignore zip codes we were not able to get the weather for.
+	weather, err := weather.GetGeo(zt.Latitude, zt.Longitude)
+	if err != nil {
+		return
+	}
+
+	zt.Temperature = weather.Temp
+	zt.Time = time.Now()
+
+	ztchan <- zt
+}

concurrency/cmd/csp/zipcodes-50.csv

@@ -0,0 +1,50 @@
+15336,40.260141,-79.992678,Gastonville,PA,Washington
+64475,40.439298,-94.62247,Parnell,MO,Nodaway
+62864,38.293269,-88.892273,Mount Vernon,IL,Jefferson
+36049,31.754655,-86.291595,Luverne,AL,Crenshaw
+30549,34.10601,-83.589021,Jefferson,GA,Jackson
+49235,41.930577,-84.156209,Clayton,MI,Lenawee
+21561,39.538568,-79.190791,Swanton,MD,Garrett
+13165,42.892556,-76.882263,Waterloo,NY,Seneca
+36473,31.466945,-87.044687,Range,AL,Conecuh
+98181,47.432251,-121.803388,Seattle,WA,King
+54874,46.568809,-91.935156,South Range,WI,Douglas
+45840,41.015219,-83.664786,Findlay,OH,Hancock
+97303,44.992731,-123.01672,Salem,OR,Marion
+91709,34.839964,-115.967051,Chino Hills,CA,San Bernardino
+68048,40.971961,-96.001244,Plattsmouth,NE,Cass
+15327,39.764089,-79.937419,Dilliner,PA,Greene
+38954,33.445684,-89.910076,Sidon,MS,Carroll
+29581,34.000746,-79.102314,Nichols,SC,Horry
+81504,39.123554,-108.589414,Grand Junction,CO,Mesa
+39161,32.498294,-89.868563,Sandhill,MS,Rankin
+85737,32.435158,-110.984499,Tucson,AZ,Pima
+55365,45.200875,-93.888099,Monticello,MN,Wright
+99176,47.12196,-117.433708,Thornton,WA,Whitman
+72532,36.061757,-91.607603,Evening Shade,AR,Sharp
+72145,35.277393,-91.732556,Searcy,AR,White
+75662,32.410306,-94.793392,Kilgore,TX,Gregg
+23272,37.524246,-77.493157,Richmond,VA,Richmond City
+99023,47.557733,-117.183637,Mica,WA,Spokane
+77802,30.620983,-96.319377,Bryan,TX,Brazos
+05463,44.878094,-73.338597,Isle La Motte,VT,Grand Isle
+27110,36.027482,-80.20728,Winston Salem,NC,Forsyth
+33404,26.645895,-80.430269,West Palm Beach,FL,Palm Beach
+84141,40.668068,-111.908297,Salt Lake City,UT,Salt Lake
+32867,28.566338,-81.260818,Orlando,FL,Orange
+78461,27.777,-97.463213,Corpus Christi,TX,Nueces
+84123,40.65963,-111.919269,Salt Lake City,UT,Salt Lake
+76299,33.20743,-97.116282,Roanoke,TX,Denton
+24830,37.375246,-81.653889,Elbert,WV,Mcdowell
+61914,39.638,-88.75367,Bethany,IL,Moultrie
+13089,43.02143,-76.197701,Liverpool,NY,Onondaga
+03774,44.077803,-72.014608,North Haverhill,NH,Grafton
+01098,42.40494,-72.896133,Worthington,MA,Hampshire
+14517,42.591883,-77.894474,Nunda,NY,Livingston
+16743,41.881851,-78.457878,Port Allegany,PA,Mckean
+10048,40.71254,-74.013289,New York,NY,New York
+19375,39.901177,-75.750203,Unionville,PA,Chester
+44608,40.665824,-81.592696,Beach City,OH,Stark
+14883,42.203102,-76.471165,Spencer,NY,Tioga
+34218,27.499515,-82.709871,Holmes Beach,FL,Manatee
+75037,32.767268,-96.777626,Irving,TX,Dallas

concurrency/cmd/mutex/main.go

@@ -0,0 +1,45 @@
+package main
+
+import (
+	"fmt"
+	"sync"
+)
+
+func main() {
+	var balance int = 100
+	// var mu sync.Mutex
+
+	fmt.Println("Initial balance:", balance)
+
+	// Apply concurrent adjustments to the balance.
+	// The sum of these adjustments is 0.
+	// We should see no changes in balance at the end.
+	var adjustments = []int{
+		-8, -4, -2, 0, 2, 4, 8,
+		-8, -4, -2, 0, 2, 4, 8,
+		-8, -4, -2, 0, 2, 4, 8,
+		-8, -4, -2, 0, 2, 4, 8,
+		-8, -4, -2, 0, 2, 4, 8,
+		-8, -4, -2, 0, 2, 4, 8,
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(len(adjustments))
+
+	for _, adjustment := range adjustments {
+		go func(adjustment int) {
+			defer wg.Done()
+
+			// mu.Lock()
+
+			balance += adjustment
+
+			// mu.Unlock()
+
+		}(adjustment)
+	}
+
+	wg.Wait()
+
+	fmt.Println("Final balance:", balance)
+}

concurrency/cmd/mutex/run.sh

@@ -0,0 +1,7 @@
+#!/bin/bash
+
+for (( ; ; )) do
+    go run ./main.go
+    printf "\n"
+    sleep 0.5
+done

concurrency/cmd/temperatures/main.go

@@ -0,0 +1,40 @@
+package main
+
+import (
+	"fmt"
+	"time"
+)
+
+var cities = []string{
+	"Chicago",
+	"Dallas",
+	"Houston",
+	"Los Angeles",
+	"New York",
+	"Philadelphia",
+	"Phoenix",
+	"San Antonio",
+	"San Diego",
+	"San Jose",
+}
+
+func main() {
+	now := time.Now()
+
+	printTemperatures(cities)
+
+	fmt.Printf(
+		"\nTook %v to get temperatures sequentially.\n\n",
+		time.Since(now),
+	)
+
+	// Reset the timer.
+	now = time.Now()
+
+	printTemperaturesAsync(cities)
+
+	fmt.Printf(
+		"\nTook %v to get temperatures concurrently.\n\n",
+		time.Since(now),
+	)
+}

concurrency/cmd/temperatures/print.go

@@ -0,0 +1,59 @@
+package main
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/dwrz/talks/concurrency/pkg/weather"
+)
+
+const (
+	// Hardcoded padding for longest city name, Philadelphia.
+	message = "%-12s %.2f°F\n"
+)
+
+func printTemperatures(cities []string) {
+	for _, city := range cities {
+		cityWeather, err := weather.GetCity(city)
+		if err != nil {
+			fmt.Printf(
+				"error getting weather for %s: %v\n", city, err,
+			)
+			continue
+		}
+
+		fmt.Printf(message, city, cityWeather.Temp)
+	}
+}
+
+func printTemperaturesAsync(cities []string) {
+	var weatherReports = make(chan string, len(cities))
+	var wg sync.WaitGroup
+	wg.Add(len(cities))
+
+	for _, city := range cities {
+		go func(city string) {
+			defer wg.Done()
+
+			cityWeather, err := weather.GetCity(city)
+			if err != nil {
+				fmt.Printf(
+					"error getting weather for %s: %v\n",
+					city, err,
+				)
+				return
+			}
+
+			weatherReports <- fmt.Sprintf(
+				message, city, cityWeather.Temp,
+			)
+		}(city)
+	}
+
+	wg.Wait()
+	close(weatherReports)
+
+	for report := range weatherReports {
+		fmt.Print(report)
+	}
+}

concurrency/cmd/temperatures/run.sh

@@ -0,0 +1,5 @@
+#!/bin/bash
+
+export OPEN_WEATHER_MAP_API_KEY="$(pass open-weather-map/api-key)"
+
+go run ./*.go