Tutorial Go: Selectores y Mutexes Concurrencia y Exclusión Mutua

Uso de select

En Go, el uso del select permite gestionar múltiples operaciones de canal de forma concurrente, lo que es esencial para evitar bloqueos. El select actúa de manera similar a una instrucción switch, pero está diseñado específicamente para trabajar con canales. Esto significa que puedes esperar en múltiples operaciones de envío o recepción al mismo tiempo, y el select elegirá una de ellas que esté lista para proceder.

El funcionamiento básico del select es evaluar múltiples casos de canal. Cada caso corresponde a una operación de envío o recepción. Si más de un caso está listo, el select elige uno al azar, lo que ayuda a distribuir la carga de trabajo de manera uniforme. Si ningún caso está listo, el select se bloqueará hasta que al menos uno lo esté, a menos que exista un caso default, que se ejecutará inmediatamente si está presente.

package main

import (
    "fmt"
    "time"
)

func main() {
    ch1 := make(chan string)
    ch2 := make(chan string)

    go func() {
        time.Sleep(1 * time.Second)
        ch1 <- "mensaje de ch1"
    }()

    go func() {
        time.Sleep(2 * time.Second)
        ch2 <- "mensaje de ch2"
    }()

    select {
    case msg1 := <-ch1:
        fmt.Println("Recibido:", msg1)
    case msg2 := <-ch2:
        fmt.Println("Recibido:", msg2)
    }

}

Salida esperada:

Recibido: mensaje de ch1

Recibido: Mensaje único

En este ejemplo, dos goroutines envían mensajes a través de dos canales diferentes, ch1 y ch2, después de un tiempo específico. El select espera a que uno de los dos canales reciba un mensaje. Importante: el uso del select asegura que el programa no se bloqueará esperando en un solo canal, sino que responderá al primero que esté listo.

Además, select permite implementar timeouts para operaciones que podrían tardar demasiado. Esto se logra utilizando el paquete time y creando un caso adicional en el select:

select {
case msg1 := <-ch1:
    fmt.Println("Mensaje recibido:", msg1)
case <-time.After(5 * time.Second):
    fmt.Println("Timeout: no se recibió ningún mensaje en 5 segundos")
}

Salida esperada:

Recibido: mensaje de ch1

En este código, si no se recibe ningún mensaje en el canal dentro de los 5 segundos, el caso del timeout se ejecuta. Esto es útil para garantizar que el programa no se bloquee indefinidamente esperando una operación que podría no completarse.

El caso default en un select se ejecuta inmediatamente si ninguno de los demás casos está listo. Esto es útil para evitar bloqueos cuando se desea realizar otras tareas si no hay canales listos:

select {
case msg1 := <-ch1:
    fmt.Println("Mensaje recibido:", msg1)
default:
    fmt.Println("No hay mensajes disponibles; realizando otras tareas")
}

Salida esperada:

Recibido: mensaje de ch1

Este ejemplo demuestra cómo el caso default permite que el flujo de ejecución continúe sin esperar indefinidamente a que el canal ch1 reciba un mensaje.

El select también es fundamental para implementar patrones avanzados de concurrencia, como multiplexación de canales y gestión de cancelaciones. Al permitir que una goroutine escuche múltiples canales, se puede diseñar un sistema que responda de manera dinámica a diferentes eventos y condiciones.

Uso de mutexes

En Go, la sincronización explícita es crucial cuando múltiples goroutines acceden a variables compartidas. 

Las condiciones de carrera ocurren cuando dos o más goroutines intentan leer y escribir en una variable al mismo tiempo, lo que puede llevar a comportamientos indeterminados y errores difíciles de rastrear. Para prevenir estas condiciones, Go proporciona la estructura sync.Mutex, que permite la exclusión mutua en el acceso a recursos compartidos.

Un Mutex es un mecanismo de bloqueo que asegura que solo una goroutine pueda acceder a una sección crítica de código a la vez. Al usar un Mutex, se deben seguir dos operaciones fundamentales: Lock y Unlock. La operación Lock bloquea el acceso a la sección crítica, y Unlock libera este bloqueo, permitiendo que otras goroutines accedan al recurso.

package main

import (
    "fmt"
    "sync"
)

var (
    contador int
    mutex    sync.Mutex
)

func incrementar(wg *sync.WaitGroup) {
    defer wg.Done()
    mutex.Lock()
    contador++
    mutex.Unlock()
}

func main() {
    var wg sync.WaitGroup
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go incrementar(&wg)
    }
    wg.Wait()
    fmt.Println("Valor final del contador:", contador)
}

En este ejemplo, se define una variable contador que es incrementada por múltiples goroutines. El uso del Mutex garantiza que solo una goroutine pueda modificar el valor de contador a la vez, evitando condiciones de carrera.

Importante: siempre que se use un mutex, es esencial asegurarse de que cada Lock() tenga un correspondiente Unlock() para evitar bloqueos permanentes. Una práctica común es usar defer inmediatamente después de Lock() para garantizar que Unlock() se llame cuando la función regrese, incluso si ocurre un error.

Además, Go ofrece el tipo sync.RWMutex, que permite un mayor control sobre el acceso concurrente a los recursos. Un RWMutex proporciona dos tipos de bloqueos: RLock para operaciones de solo lectura y Lock para operaciones de escritura. Esto permite que múltiples goroutines puedan leer simultáneamente de un recurso, pero asegura que solo una goroutine pueda escribir en él.

package main

import (
    "fmt"
    "sync"
)

var (
    contador   int
    rwMutex    sync.RWMutex
)

func leerContador() int {
    rwMutex.RLock()
    defer rwMutex.RUnlock()
    return contador
}

func incrementarContador(wg *sync.WaitGroup) {
    defer wg.Done()
    rwMutex.Lock()
    contador++
    rwMutex.Unlock()
}

func main() {
    var wg sync.WaitGroup
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go incrementarContador(&wg)
    }
    wg.Wait()
    fmt.Println("Valor final del contador:", leerContador())
}

En este segundo ejemplo, se utiliza sync.RWMutex para permitir lecturas concurrentes mientras se asegura que las escrituras sean exclusivas. Esto es útil en escenarios donde las operaciones de lectura son más frecuentes que las de escritura, mejorando así el rendimiento sin comprometer la seguridad de los datos.

El uso de mutexes es fundamental para evitar las condiciones de carrera, que pueden dar lugar a resultados inesperados o erróneos en programas concurrentes. Garantizan que las operaciones sobre variables compartidas sean atómicas y seguras, lo que es esencial para mantener la consistencia de los datos en un entorno concurrente.

Ejemplos de patrones concurrentes

En aplicaciones concurrentes desarrolladas en Go, es común combinar el uso de selectores y mutexes para manejar operaciones en canales y proteger datos compartidos, respectivamente. 

Estos dos mecanismos se utilizan para abordar diferentes aspectos de la concurrencia: 

• el select para elegir entre múltiples operaciones de canal
• los mutexes para garantizar la exclusión mutua en el acceso a recursos compartidos.

Patrón de multiplexación de canales con protección de datos compartidos

Supongamos que tenemos múltiples fuentes de datos que alimentan a una variable compartida, y necesitamos actualizar esta variable sin generar condiciones de carrera. Podemos utilizar un select para escuchar múltiples canales y un mutex para sincronizar el acceso a la variable compartida.

package main

import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)

var (
    valorCompartido int
    mutex           sync.Mutex
)

func generarDatos(canal chan int, id int) {
    for {
        tiempoEspera := time.Duration(rand.Intn(1000)) * time.Millisecond
        time.Sleep(tiempoEspera)
        dato := rand.Intn(100)
        fmt.Printf("Goroutine %d generó dato %d\n", id, dato)
        canal <- dato
    }
}

func main() {
    canal1 := make(chan int)
    canal2 := make(chan int)

    go generarDatos(canal1, 1)
    go generarDatos(canal2, 2)

    for {
        select {
        case dato := <-canal1:
            mutex.Lock()
            valorCompartido += dato
            fmt.Printf("Valor actualizado desde canal1: %d\n", valorCompartido)
            mutex.Unlock()
        case dato := <-canal2:
            mutex.Lock()
            valorCompartido += dato
            fmt.Printf("Valor actualizado desde canal2: %d\n", valorCompartido)
            mutex.Unlock()
        }
    }
}

En este ejemplo, dos goroutines generan datos aleatorios y los envían a través de canales independientes. El select en el main escucha ambos canales y, cuando recibe datos, bloquea el acceso a valorCompartido utilizando el mutex, actualiza su valor y luego libera el mutex. De esta manera, se garantiza la consistencia del dato compartido pese al acceso concurrente.

Patrón de trabajador concurrente con canal de tareas y mutex de estado

Otra situación común es la necesidad de distribuir tareas entre varios trabajadores concurrentes y administrar el estado compartido. Aquí se muestra cómo implementar un pool de trabajadores que procesan tareas de un canal y actualizan un contador compartido protegido por un mutex.

package main

import (
    "fmt"
    "sync"
)

var (
    contadorExitos int
    mutex          sync.Mutex
    wg             sync.WaitGroup
)

func trabajador(id int, tareas <-chan int) {
    defer wg.Done()
    for tarea := range tareas {
        // Procesamiento de la tarea
        fmt.Printf("Trabajador %d procesando tarea %d\n", id, tarea)
        // Simular éxito en el procesamiento
        mutex.Lock()
        contadorExitos++
        mutex.Unlock()
    }
}

func main() {
    numTrabajadores := 3
    tareas := make(chan int, 10)

    for i := 1; i <= numTrabajadores; i++ {
        wg.Add(1)
        go trabajador(i, tareas)
    }

    for i := 1; i <= 20; i++ {
        tareas <- i
    }
    close(tareas)

    wg.Wait()
    fmt.Printf("Número total de tareas exitosas: %d\n", contadorExitos)
}

En este código, se crean varias goroutines como trabajadores que reciben tareas de un canal. Cada trabajador procesa una tarea y actualiza un contador compartido. El acceso al contador compartido está protegido por un mutex para evitar condiciones de carrera. Este patrón es útil para aprovechar múltiples núcleos y mejorar el desempeño en operaciones intensivas.

Patrón de cancelación controlada con select y mutexes

En ocasiones, es necesario poder cancelar operaciones concurrentes de manera segura. Utilizando un canal de señalización y un select, podemos coordinar la terminación de goroutines y asegurar que los recursos compartidos se gestionen adecuadamente con mutexes.

package main

import (
    "fmt"
    "sync"
    "time"
)

var (
    contador int
    mutex    sync.Mutex
)

func proceso(id int, cancel <-chan struct{}, wg *sync.WaitGroup) {
    defer wg.Done()
    for {
        select {
        case <-cancel:
            fmt.Printf("Proceso %d cancelado\n", id)
            return
        default:
            // Trabajo simulado
            time.Sleep(500 * time.Millisecond)
            mutex.Lock()
            contador++
            fmt.Printf("Proceso %d incrementó contador a %d\n", id, contador)
            mutex.Unlock()
        }
    }
}

func main() {
    cancel := make(chan struct{})
    var wg sync.WaitGroup

    for i := 1; i <= 3; i++ {
        wg.Add(1)
        go proceso(i, cancel, &wg)
    }

    time.Sleep(2 * time.Second)
    close(cancel) // Señal de cancelación

    wg.Wait()
    fmt.Println("Todos los procesos han terminado")
}

En este ejemplo, múltiples goroutines ejecutan un proceso que incrementa un contador compartido. Utilizando un canal de cancelación, podemos indicar a las goroutines que deben terminar su ejecución. El select dentro de cada goroutine verifica si se ha recibido la señal de cancelación y, en tal caso, finaliza de manera controlada. El acceso al contador está protegido por un mutex, asegurando la integridad de los datos hasta el final.

Patrón de productor-consumidor con buffers y sincronización

Una variación común es implementar el patrón productor-consumidor, donde uno o más productores generan datos que son consumidos por uno o más consumidores. El uso de buffers y la sincronización adecuada es crucial para evitar bloqueos y condiciones de carrera.

package main

import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)

func main() {
    buffer := make(chan int, 10)
    var wgProductores, wgConsumidores sync.WaitGroup

    // Iniciar productores
    for i := 1; i <= 2; i++ {
        wgProductores.Add(1)
        go productor(i, buffer, &wgProductores)
    }

    // Iniciar consumidores
    for i := 1; i <= 3; i++ {
        wgConsumidores.Add(1)
        go consumidor(i, buffer, &wgConsumidores)
    }

    wgProductores.Wait()
    close(buffer)
    wgConsumidores.Wait()
    fmt.Println("Todos los productores y consumidores han terminado")
}

func productor(id int, buffer chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for i := 0; i < 5; i++ {
        tiempoEspera := time.Duration(rand.Intn(500)) * time.Millisecond
        time.Sleep(tiempoEspera)
        dato := rand.Intn(100)
        buffer <- dato
        fmt.Printf("Productor %d produjo dato %d\n", id, dato)
    }
}

func consumidor(id int, buffer <-chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    for dato := range buffer {
        fmt.Printf("Consumidor %d recibió dato %d\n", id, dato)
        // Procesar dato...
    }
}

En este código, los productores generan datos y los envían a través de un canal con buffer. Los consumidores reciben los datos del canal y los procesan. El canal con buffer actúa como un almacén que regula el flujo entre productores y consumidores, evitando bloqueos cuando los consumidores no pueden procesar datos inmediatamente. Aunque en este ejemplo no se hace uso de mutexes, si hubiera una variable compartida entre productores y consumidores, se debería proteger con un mutex.

Patrón de uso de sync.Cond para coordinación avanzada

Para casos donde se requiere una coordinación más sofisticada entre goroutines, Go proporciona sync.Cond, que permite que una goroutine espere hasta que se cumpla una condición determinada, mientras otra goroutine la notifica cuando esa condición cambia. Esto es útil para implementar patrones de espera y señalización.

package main

import (
    "fmt"
    "sync"
)

type almacen struct {
    datos []int
    mutex sync.Mutex
    cond  *sync.Cond
}

func nuevoAlmacen() *almacen {
    a := &almacen{}
    a.cond = sync.NewCond(&a.mutex)
    return a
}

func (a *almacen) producir(dato int) {
    a.mutex.Lock()
    a.datos = append(a.datos, dato)
    fmt.Printf("Producido: %d\n", dato)
    a.mutex.Unlock()
    a.cond.Signal()
}

func (a *almacen) consumir() {
    a.mutex.Lock()
    for len(a.datos) == 0 {
        a.cond.Wait()
    }
    dato := a.datos[0]
    a.datos = a.datos[1:]
    fmt.Printf("Consumido: %d\n", dato)
    a.mutex.Unlock()
}

func main() {
    a := nuevoAlmacen()
    var wg sync.WaitGroup

    // Consumidor
    wg.Add(1)
    go func() {
        defer wg.Done()
        for i := 0; i < 5; i++ {
            a.consumir()
        }
    }()

    // Productor
    for i := 1; i <= 5; i++ {
        a.producir(i)
    }

    wg.Wait()
    fmt.Println("Proceso completado")
}

En este ejemplo, implementamos un almacén donde un consumidor espera hasta que haya datos disponibles. Utilizando sync.Cond, el consumidor puede esperar de manera eficiente sin consumir recursos de CPU, y el productor puede notificar cuando hay nuevos datos. El uso combinado de mutex y condicionales proporciona una manera robusta de coordinar acciones entre múltiples goroutines.

El uso combinado de selectores y mutexes aseguran que las goroutines respondan a eventos externos de manera concurrente mientras protegen el acceso a los datos compartidos, evitando condiciones de carrera. Este enfoque es particularmente útil en sistemas donde el flujo de datos y las señales de control deben gestionarse de manera coordinada.