performance - ¿Por qué estas gorutinas no escalan su rendimiento a partir de ejecuciones más concurrentes?

optimization concurrency

Question

Fondo

Actualmente estoy trabajando en mi tesis de licenciatura y básicamente mi tarea es optimizar un código dado en Go, es decir, hacer que se ejecute lo más rápido posible. Primero, optimicé la función en serie y luego intenté introducir el paralelismo a través de goroutines. Después de investigar en Internet, ahora entiendo la diferencia entre concurrencia y paralelismo gracias a las siguientes diapositivas de talks.golang . Visité algunos cursos de programación paralela donde comparamos el código ac / c ++ con la ayuda de pthread / openmp, por lo que traté de aplicar estos paradigmas en Go. Dicho esto, en este caso particular, estoy optimizando una función que calcula el promedio móvil de un segmento con longitud len:=n+(window_size-1) (es igual a 9393 o 10175), por lo tanto tenemos n ventanas de las cuales calculamos el promedio aritmético correspondiente y guárdelo correctamente en el segmento de salida.

Tenga en cuenta que esta tarea es inherentemente paralela vergonzosa.

Mis intentos de optimización y resultados

En moving_avg_concurrent2 el segmento en num_goroutines piezas más pequeñas y ejecuté cada una con una goroutine. Esta función se realizó con una goroutina, por alguna razón (no se pudo encontrar por qué todavía, pero aquí nos estamos moving_avg_serial4 tangentes), mejor que moving_avg_serial4 pero con más de una goroutine comenzó a funcionar peor que moving_avg_serial4 .
En moving_avg_concurrent3 adopté el paradigma maestro / trabajador. El rendimiento fue peor que moving_avg_serial4 cuando se usaba una rutina. Aquí, al menos, obtuve un mejor rendimiento al aumentar num_goroutines pero aún no es mejor que moving_avg_serial4 . Para comparar el rendimiento de moving_avg_serial4 , moving_avg_concurrent2 y moving_avg_concurrent3 escribí un punto de referencia y tabulé los resultados:

fct & num_goroutines | timing in ns/op | percentage  
---------------------------------------------------------------------   
          serial4    |         4357893 |   100.00%  
          concur2_1  |         5174818 |   118.75%  
          concur2_4  |         9986386 |   229.16%  
          concur2_8  |        18973443 |   435.38%  
          concur2_32 |        75602438 |  1734.84%  
          concur3_1  |        32423150 |   744.01%  
          concur3_4  |        21083897 |   483.81%  
          concur3_8  |        16427430 |   376.96%  
          concur3_32 |        15157314 |   347.81%  

Pregunta

Como, como se mencionó anteriormente, este problema es vergonzosamente paralelo, esperaba ver un enorme aumento del rendimiento, pero ese no fue el caso.

¿Por qué moving_avg_concurrent2 no escala en absoluto?
¿Y por qué moving_avg_concurrent3 es mucho más lento que moving_avg_serial4 ?
Sé que las gorutinas son baratas pero aún no lo son, pero ¿es posible que esto genere tanta sobrecarga que sea aún más lenta que moving_avg_serial4 ?

Código

Funciones:

// returns a slice containing the moving average of the input (given, i.e. not optimised)
func moving_avg_serial(input []float64, window_size int) []float64 {
    first_time := true
    var output = make([]float64, len(input))
    if len(input) > 0 {
        var buffer = make([]float64, window_size)
        // initialise buffer with NaN
        for i := range buffer {
            buffer[i] = math.NaN()
        }
        for i, val := range input {
            old_val := buffer[int((math.Mod(float64(i), float64(window_size))))]
            buffer[int((math.Mod(float64(i), float64(window_size))))] = val
            if !NaN_in_slice(buffer) && first_time {
                sum := 0.0
                for _, entry := range buffer {
                    sum += entry
                }
                output[i] = sum / float64(window_size)
                first_time = false
            } else if i > 0 && !math.IsNaN(output[i-1]) && !NaN_in_slice(buffer) {
                output[i] = output[i-1] + (val-old_val)/float64(window_size) // solution without loop
            } else {
                output[i] = math.NaN()
            }
        }
    } else { // empty input
        fmt.Println("moving_avg is panicking!")
        panic(fmt.Sprintf("%v", input))
    }
    return output
}

// returns a slice containing the moving average of the input
// reordering the control structures to exploid the short-circuit evaluation
func moving_avg_serial4(input []float64, window_size int) []float64 {
    first_time := true
    var output = make([]float64, len(input))
    if len(input) > 0 {
        var buffer = make([]float64, window_size)
        // initialise buffer with NaN
        for i := range buffer {
            buffer[i] = math.NaN()
        }
        for i := range input {
            //            fmt.Printf("in mvg_avg4: i=%v/n", i)
            old_val := buffer[int((math.Mod(float64(i), float64(window_size))))]
            buffer[int((math.Mod(float64(i), float64(window_size))))] = input[i]
            if first_time && !NaN_in_slice(buffer) {
                sum := 0.0
                for j := range buffer {
                    sum += buffer[j]
                }
                output[i] = sum / float64(window_size)
                first_time = false
            } else if i > 0 && !math.IsNaN(output[i-1]) /* && !NaN_in_slice(buffer)*/ {
                output[i] = output[i-1] + (input[i]-old_val)/float64(window_size) // solution without loop
            } else {
                output[i] = math.NaN()
            }
        }
    } else { // empty input
        fmt.Println("moving_avg is panicking!")
        panic(fmt.Sprintf("%v", input))
    }
    return output
}

// returns a slice containing the moving average of the input
// splitting up slice into smaller pieces for the goroutines but without using the serial version, i.e. we only have NaN''s in the beginning, thus hope to reduce some overhead
// still does not scale (decreasing performance with increasing size and num_goroutines)
func moving_avg_concurrent2(input []float64, window_size, num_goroutines int) []float64 {
    var output = make([]float64, window_size-1, len(input))
    for i := 0; i < window_size-1; i++ {
        output[i] = math.NaN()
    }
    if len(input) > 0 {
        num_items := len(input) - (window_size - 1)
        var barrier_wg sync.WaitGroup
        n := num_items / num_goroutines
        go_avg := make([][]float64, num_goroutines)
        for i := 0; i < num_goroutines; i++ {
            go_avg[i] = make([]float64, 0, num_goroutines)
        }

        for i := 0; i < num_goroutines; i++ {
            barrier_wg.Add(1)
            go func(go_id int) {
                defer barrier_wg.Done()

                // computing boundaries
                var start, stop int
                start = go_id*int(n) + (window_size - 1) // starting index
                // ending index
                if go_id != (num_goroutines - 1) {
                    stop = start + n // Ending index
                } else {
                    stop = num_items + (window_size - 1) // Ending index
                }

                loc_avg := moving_avg_serial4(input[start-(window_size-1):stop], window_size)

                loc_avg = make([]float64, stop-start)
                current_sum := 0.0
                for i := start - (window_size - 1); i < start+1; i++ {
                    current_sum += input[i]
                }
                loc_avg[0] = current_sum / float64(window_size)
                idx := 1

                for i := start + 1; i < stop; i++ {
                    loc_avg[idx] = loc_avg[idx-1] + (input[i]-input[i-(window_size)])/float64(window_size)
                    idx++
                }

                go_avg[go_id] = append(go_avg[go_id], loc_avg...)

            }(i)
        }
        barrier_wg.Wait()

        for i := 0; i < num_goroutines; i++ {
            output = append(output, go_avg[i]...)
        }

    } else { // empty input
        fmt.Println("moving_avg is panicking!")
        panic(fmt.Sprintf("%v", input))
    }
    return output
}

// returns a slice containing the moving average of the input
// change of paradigm, we opt for a master worker pattern and spawn all windows which each will be computed by a goroutine
func compute_window_avg(input, output []float64, start, end int) {
    sum := 0.0
    size := end - start
    for _, val := range input[start:end] {
        sum += val
    }
    output[end-1] = sum / float64(size)
}

func moving_avg_concurrent3(input []float64, window_size, num_goroutines int) []float64 {
    var output = make([]float64, window_size-1, len(input))
    for i := 0; i < window_size-1; i++ {
        output[i] = math.NaN()
    }
    if len(input) > 0 {
        num_windows := len(input) - (window_size - 1)
        var output = make([]float64, len(input))
        for i := 0; i < window_size-1; i++ {
            output[i] = math.NaN()
        }

        pending := make(chan *Work)
        done := make(chan *Work)

        // creating work
        go func() {
            for i := 0; i < num_windows; i++ {
                pending <- NewWork(compute_window_avg, input, output, i, i+window_size)
            }
        }()

        // start goroutines which work through pending till there is nothing left
        for i := 0; i < num_goroutines; i++ {
            go func() {
                Worker(pending, done)
            }()
        }

        // wait till every work is done
        for i := 0; i < num_windows; i++ {
            <-done
        }

        return output

    } else { // empty input
        fmt.Println("moving_avg is panicking!")
        panic(fmt.Sprintf("%v", input))
    }
    return output
}

Puntos de referencia:

//############### BENCHMARKS ###############
var import_data_res11 []float64
func benchmarkMoving_avg_serial(b *testing.B, window int) {
    var r []float64
    for n := 0; n < b.N; n++ {
        r = moving_avg_serial(BackTest_res.F["Trading DrawDowns"], window)
    }
    import_data_res11 = r
}

var import_data_res14 []float64
func benchmarkMoving_avg_serial4(b *testing.B, window int) {
    var r []float64
    for n := 0; n < b.N; n++ {
        r = moving_avg_serial4(BackTest_res.F["Trading DrawDowns"], window)
    }
    import_data_res14 = r
}

var import_data_res16 []float64
func benchmarkMoving_avg_concurrent2(b *testing.B, window, num_goroutines int) {
    var r []float64
    for n := 0; n < b.N; n++ {
        r = moving_avg_concurrent2(BackTest_res.F["Trading DrawDowns"], window, num_goroutines)
    }
    import_data_res16 = r
}

var import_data_res17 []float64
func benchmarkMoving_avg_concurrent3(b *testing.B, window, num_goroutines int) {
    var r []float64
    for n := 0; n < b.N; n++ {
        r = moving_avg_concurrent3(BackTest_res.F["Trading DrawDowns"], window, num_goroutines)
    }
    import_data_res17 = r
}



func BenchmarkMoving_avg_serial_261x10(b *testing.B) {
    benchmarkMoving_avg_serial(b, 261*10)
}

func BenchmarkMoving_avg_serial4_261x10(b *testing.B) {
    benchmarkMoving_avg_serial4(b, 261*10)
}


func BenchmarkMoving_avg_concurrent2_261x10_1(b *testing.B) {
    benchmarkMoving_avg_concurrent2(b, 261*10, 1)
}
func BenchmarkMoving_avg_concurrent2_261x10_8(b *testing.B) {
    benchmarkMoving_avg_concurrent2(b, 261*10, 8)
}


func BenchmarkMoving_avg_concurrent3_261x10_1(b *testing.B) {
    benchmarkMoving_avg_concurrent3(b, 261*10, 1)
}
func BenchmarkMoving_avg_concurrent3_261x10_8(b *testing.B) {
    benchmarkMoving_avg_concurrent3(b, 261*10, 8)
}
//############### BENCHMARKS end ###############

Observaciones:
Esta es mi primera publicación, todavía estoy aprendiendo, por lo que cualquier crítica constructiva también es bienvenida.

Answer 1

Hecho # 0: los esfuerzos de optimización prematuros a menudo tienen rendimientos negativos
_{mostrando que son solo una pérdida de tiempo y esfuerzo}

¿Por qué?
Un solo SLOC "incorrecto" puede devastar el rendimiento en más de aproximadamente + 37%
o puede mejorar el rendimiento para gastar menos del -57% del tiempo de procesamiento de referencia

51.151µs on MA(200) [10000]float64    ~ 22.017µs on MA(200) [10000]int
70.325µs on MA(200) [10000]float64

¿Por qué []int -s?
Puede verlo usted mismo arriba: este es el pan y la mantequilla para las estrategias de procesamiento de sub-us [HP] eficientes de fintech (y todavía hablamos en términos de programación de procesos [SERIAL] ).

Este puede probar en cualquier escala, pero más bien pruebe primero (aquí) sus propias implementaciones, en la misma escala MA(200) [10000]float64 configuración MA(200) [10000]float64 ) y publique las duraciones de línea de base en [us] para ver el proceso inicial rendimiento y comparar manzanas con manzanas , teniendo el umbral publicado 51.2 [us] para comparar.

Luego viene la parte más difícil:

Hecho # 1: Esta tarea NO es vergonzosamente paralela

Sí, uno puede ir e implementar un cálculo de la Media Móvil, de modo que de hecho proceda a través de los montones de datos utilizando un enfoque de procesamiento "justo" - [CONCURRENT] intencionalmente adoctrinado (independientemente de si se debe a algún tipo de error, el consejo de alguna autoridad ) " , ceguera profesional o simplemente por una ignorancia doble de Sócrates, que obviamente no significa que la naturaleza del procesamiento de flujo convolucional, presente dentro de la formulación matemática de la Media Móvil, se haya olvidado de ser un proceso puro [SERIAL] , solo debido a un intento de hacer que se calcule dentro de cierto grado de procesamiento "justo" - [CONCURRENT] .

(Por cierto, los Hard Computer-Scientists y los nerds de doble dominio también objetarán aquí, que Go-language es, por diseño, utilizando las mejores habilidades de Rob Pike para tener un marco de rutinas concurrentes, no ninguna verdadera programación de procesos [PARALLEL] , incluso aunque las herramientas CSP de Hoare, disponibles en el concepto de lenguaje, pueden agregar algo de sal y pimienta e introducir un tipo de bloqueo de herramientas de comunicación entre procesos, que bloqueará secciones de código "simplemente" - [CONCURRENT] en algunos CSP-p2p cableados -sincronización.)

Hecho # 2: Ve distribuido (para cualquier tipo de aceleración) solo AL FINAL

Tener un bajo nivel de rendimiento en [SERIAL] no establece ningún criterio. Con una cantidad razonable de ajuste de rendimiento en un solo hilo, solo entonces uno puede beneficiarse de distribuirse (aún así tener que pagar costos de serie adicionales, lo que hace que la Ley Amdahl (en lugar de la estricta Ley de Amdahl ) ingrese al juego).

Si se puede introducir un nivel tan bajo de gastos generales de configuración adicionales y aún lograr un paralelismo notable, escalado en la parte no [SEQ] del procesamiento , allí y solo existe la posibilidad de aumentar el rendimiento efectivo del proceso.

No es difícil perder mucho más que ganar en esto, por lo que siempre debe comparar el [SEQ] con las posibles compensaciones entre un proceso non-[SEQ] / N[PAR]_processes teórico, ingenuo por encima de la cabeza, para el cual pagará el costo de una suma de todos los gastos generales adicionales [SEQ] , por lo que si y solo si:

(         pure-[SEQ]_processing      [ns]
+       add-on-[SEQ]-setup-overheads [ns]
+        ( non-[SEQ]_processing      [ns] / N[PAR]_processes )
  ) << (  pure-[SEQ]_processing      [ns]
       + ( non-[SEQ]_processing      [ns] / 1 )
         )

Al no tener esta ventaja de los aviones de combate de la altura excedente y el Sol detrás de usted, nunca intente entrar en ningún tipo de intentos de HPC / paralelización: nunca pagarán por sí mismos sin ser notablemente << mejores que un proceso inteligente [SEQ] .

Una animación vale más que millones de palabras.

Asi que,
asumir un proceso bajo prueba, que tiene una parte [SERIAL] y una [PARALLEL] del cronograma del proceso.

Supongamos que p es la fracción [PARALLEL] de la duración del proceso ~ ( 0.0 .. 1.0 ) tanto, la parte [SERIAL] no dura más que ( 1 - p ) , ¿verdad?

Entonces, comencemos la experimentación interactiva a partir de un caso de prueba de este tipo, donde p == 1.0 , lo que significa que toda la duración del proceso se gasta solo en una parte [PARALLEL] , y las partes iniciales y finales del flujo del proceso ( que principalmente son [SERIAL] ) tienen duraciones cero ( ( 1 - p ) == 0. )

Suponga que el sistema no hace magia en particular y, por lo tanto, necesita gastar algunos pasos reales en la inicialización de cada una de las partes [PARALLEL] , para ejecutarlo en un procesador diferente ( (1), 2, .., N ) , así que vamos agregue algunos gastos generales, si se le pide que reorganice el flujo del proceso y organice + distribuya + desarme todas las instrucciones y datos necesarios, para que el proceso previsto ahora pueda comenzar y ejecutarse en N procesadores en paralelo.

Estos costos se denominan o (aquí se supone inicialmente que la simplicidad es solo constante e invariable a N , que no siempre es el caso real, en silicio / en NUMA / en infraestructuras distribuidas).

Al hacer clic en el título del Epílogo anterior, se abre un entorno interactivo y es gratuito para la propia experimentación.

Con p == 1. && o == 0. && N > 1 el rendimiento está creciendo abruptamente a los límites de O / S de hardware [PARALLEL] alcanzables actuales para una ejecución de código de O / S todavía monolítica (donde todavía no hay costos de distribución adicionales para distribuciones de unidades de trabajo en modo dependiente MPI y similares (donde uno tendría que agregar inmediatamente una gran cantidad de [ms] , mientras que nuestra mejor implementación [SERIAL] hasta ahora obviamente ha hecho todo el trabajo en menos que solo ~ 22.1 [nosotros] )).

Pero a excepción de este caso artificialmente optimista, el trabajo no parece tan barato para obtener un paralelismo eficiente.

• Intente no tener un cero, sino solo alrededor de ~ 0.01% de los costos generales de configuración de o , y la línea comienza a mostrar una naturaleza muy diferente de la escala consciente de los gastos generales incluso para el caso extremo [PARALLEL] extremo (aún p == 1.0 ), y tener la aceleración potencial en algún lugar cerca de la mitad del caso de aceleración lineal inicialmente súper idealista.

• Ahora, convierta la p en algo más cercano a la realidad, en un lugar menos artificialmente establecido que el caso súper idealista inicial de == 1.00 --> { 0.99, 0.98, 0.95 } y ... bingo, esta es la realidad, donde el proceso- la programación debe ser probada y validada previamente.

Qué significa eso?

Como ejemplo, si una sobrecarga (de lanzamiento + final uniéndose a un grupo de corutinas) tomaría más de ~ 0.1% de la duración real de la sección de procesamiento [PARALLEL] , no habría una aceleración mayor de 4x (aproximadamente un 1/4 de la duración original en el tiempo) para 5 corutinas (que tienen p ~ 0.95), no más de 10 veces (una duración 10 veces más rápida) para 20 corutinas (todo suponiendo que un sistema tenga 5 núcleos de CPU, respectivamente 20-CPU- núcleos libres y disponibles y listos (mejor con procesos / hilos mapeados por afinidad de núcleo de CPU de nivel O / S) para servir ininterrumpidamente a todas esas corutinas durante toda su vida útil, a fin de lograr cualquier aceleración esperada por encima.

Al no tener esa cantidad de recursos de hardware libres y listos para todas esas unidades de tareas, destinadas a implementar la parte [PARALLEL] del cronograma del proceso, los estados de bloqueo / espera introducirán estados de espera absolutos adicionales y el rendimiento resultante agrega estas nuevas secciones de bloqueo / espera [SERIAL] a la duración total del proceso y las aceleraciones inicialmente deseadas dejan de existir repentinamente y el factor de rendimiento cae muy por debajo de << 1.00 (lo que significa que el tiempo de ejecución efectivo era debido a los estados de bloqueo de manera más lenta, que el flujo de trabajo just- [SERIAL] no paralelizado).

Esto puede sonar complicado para los nuevos experimentadores entusiastas, sin embargo, podemos ponerlo en una perspectiva inversa. Dado todo el proceso de distribución, se sabe que el grupo de tareas [PARALLEL] previsto no es más corto que, digamos, alrededor de un 10 [us] , según muestran los gráficos estrictamente generales, debe haber al menos aproximadamente 1000 x 10 [us] de procesamiento intensivo de computación sin bloqueo dentro de la sección [PARALLEL] para no devastar la eficiencia del procesamiento en paralelo.

Si no hay una pieza de procesamiento lo suficientemente "gorda", los costos generales (que van notablemente por encima del umbral de ~ 0.1% ) devastan brutalmente la eficiencia neta del procesamiento paralelizado con éxito (pero habiendo funcionado en tal costos relativos injustificadamente altos de la configuración frente a los efectos netos limitados de cualquiera de los procesadores N , como se demostró en los gráficos en vivo disponibles).

No es sorprendente para los nerds de computación distribuida, que la sobrecarga o también conlleva dependencias adicionales: en N (cuantos más procesos, más esfuerzos se destinarán a distribuir paquetes de trabajo), en tamaños de BLOB de datos ordenados (el cuanto más grandes sean los BLOB, más tiempo permanecerán bloqueados los dispositivos MEM / IO, antes de servir el siguiente proceso para recibir un BLOB distribuido a través de dicho dispositivo / recurso para cada uno de los objetivos 2..N -en proceso de recepción), en evitar / CSP -coordinadas, coordinaciones entre procesos mediadas por canales (llámelo bloqueo adicional por incidente, reduciendo el p más y más por debajo del ideal finalmente agradable de 1. ).

Entonces, la realidad del mundo real está bastante lejos de ser inicialmente idealizada, agradable y prometedora p == 1.0 , ( 1 - p ) == 0.0 y o == 0.0

Como es obvio desde el principio, intente superar el umbral 22.1 [us] [SERIAL] , en lugar de tratar de superarlo, empeorando cada vez más, si va a [PARALLEL] donde los gastos generales y la escala realistas, utilizando enfoques que ya no funcionan bien , no ayuda ni un poco.