Categorías
C#

Reuniendo piezas de un vector

Cuando hay que utilizar instrucciones vectoriales, es siempre aconsejable que se cargue en memoria un trozo consecutivo de memoria para crear vectores. Para un programador ya entrenado en AVX/AVX2/AVX512, por ejemplo, es fácil darse cuenta de que el siguiente bucle (un producto escalar de dos vectores) es un candidato ideal para usar instrucciones SIMD:

double sum = 0.0;
for (int i = 0; i < size; i++)
    sum += p[i] * q[i];

Hay dos zonas de memoria, p y q, y el índice se mueve ascendentemente de uno en uno, en ambos casos. Por lo tanto, para convertir este código en instrucciones vectoriales de AVX, hacemos que cada paso del código cargue cuatro entradas de cada vector, en cada paso del bucle:

double sum = 0;
int i = 0;
if (Fma.IsSupported)
{
    var acc = Vector256<double>.Zero;
    for (; i < size - 4; i += 4)
        acc = Fma.MultiplyAdd(
            Avx.LoadVector256(p + i),
            Avx.LoadVector256(q + i),
            acc);
    acc = Avx.HorizontalAdd(acc, acc);
    sum = acc.ToScalar() + acc.GetElement(2);
}
for (; i < size; i++)
    sum += p[i] * q[i];

Gracias a este sencillo truco, se aprovecha mejor la caché de memoria del procesador. Es cierto que p y q pueden estar en direcciones distantes, pero al menos la carga de cada trozo de cuatro valores no provoca invalidaciones de la caché. En cambio, este otro fragmento de código, extraído del código de Austra que realiza la descomposición de una matriz según la factorización LU, no cumple con esta regla:

double m = c[k];
for (int i = k + 1; i < size; i++)
    c[i] -= m * a[i * size + k];

El almacenamiento en c es consecutivo, pero si reunimos cuatro pasos del bucle, las cargas desde el vector a van a venir de zonas dispersas de la memoria: esto es porque la variable de bucle en el indexador está multiplicada por size. Sería una pena no poder utilizar AVX aquí, porque podríamos reducir por cuatro las restas y multiplicaciones del bucle.

Por fortuna, Intel agregó al conjunto de instrucciones de AVX2 una instrucción que, precisamente, sirve para reunir en una misma carga fragmentos de memoria no consecutivos. Se trata de la instrucción vgatherdpd, que .NET implementa mediante el método Avx2.GatherVector:

public static Vector256<double> GatherVector256(
    double* baseAddress,
    Vector128<int> index,
    byte scale);

A primera vista, parece arameo antiguo. A segunda y tercera vistas, se asemeja más al sumerio de chabolas. Pero no es tan complicado explicar cómo funciona, sobre todo si lo acompañamos con un ejemplo. A este método le pasamos un puntero a un array de números de doble precisión, esa es la parte obvia. Además, le pasamos cuatro índices de tipo entero, en el parámetro de tipo Vector128<int>. Como este tipo contiene 128 bits, es evidente que ahí podemos pasar cuatro enteros de 32 bits; ya veremos cómo. ¿Y el factor de escala? Aquí es un poco redundante, y Microsoft podría habernos ahorrado el trabajo: cuando se trabaja con vectores de double, tenemos que pasar un 8, o lo que es igual, sizeof(double).

Juntando las piezas, he aquí como quedaría el código transformado del fragmento original:

double m = c[k];
int i = k + 1;
if (Avx2.IsSupported)
{
    var vm = Vector256.Create(m);
    var vx = Vector128.Create(0, size, 2 * size, 3 * size);
    for (double* p = a + i * size + k;
        i < size - 4;
        i += 4, p += size * 4)
        Avx.Store(c + i,
            Avx.Subtract(Avx.LoadVector256(c + i),
            Avx.Multiply(Avx2.GatherVector256(p, vx, 8), vm)));
}
for (; i < size; i++)
    c[i] -= m * a[i * size + k];

Expliquémonos: antes de entrar en el bucle, creamos un vector de números reales a partir del valor de la variable m. Lo que sigue es lo que nos interesa: creamos por adelantado, un Vector128<int> con cuatro offsets o desplazamientos diferentes, múltiplos del tamaño de la fila de la matriz, que viene en size. Y ahora viene el detalle artesanal delicado: al iniciar el bucle, creo una variable adicional de puntero, p, que apunte i * size + k entradas a partir de la dirección inicial, que está en a. Recuerde que leíamos el valor de a[i * size + k] en cada paso del bucle original, y que i es la variable de bucle. Si hubiese querido ser más claro, debería haber escrito:

double* p = a + (k + 1) * size + k

Pero me puede la pereza, y al fin y al cabo, en ese preciso instante, i es equivalente a k + 1, ¿no?

El detalle es que, en cada paso del bucle AVX, la variable p se incrementa en 4 * size entradas (es decir, en cuatro filas enteras de size entradas). Digo que es un detallazo elegante porque la alternativa obvia sería incrementar directamente los cuatro índices enteros que guardamos en el vector vx. ¿Por qué no lo he hecho así? Pues porque, al trabajar con vectores de 128 bits, la instrucción de suma pertenecería al conjunto de instrucciones SSE2… y resulta que el cambio entre AVX y SSE tiene un coste pequeño, pero no desdeñable.

… y este es el momento más apropiado para exclamar ¡pero coño! (con perdón), ¿esto no tendría que hacerlo el compilador, o el JIT, o quien narices sea?. Pues sí. De hecho, teóricamente, Hotspot, el JIT de Java, tendría que ocuparse de estas cosas. En la práctica, la vectorización automática en Hotspot es un truño como un puño. Y el famoso escape analysis tampoco hace lo que se supone que tendría que hacer. O si lo hace, no se nota. Al final, Java ha añadido una librería de vectores basados en instrucciones SIMD, al estilo de .NET. Los compiladores de C/C++ hacen mejor papel en estos casos, pero no son omnipotentes, y a veces tienes que echarles una mano.

NOTA: AVX2, creo recordar, se introdujo en la cuarta generación de Intel, Haswell, en 2013. La primera implementación en hardware de vgatherdpd no era para tirar cohetes, y todavía hay páginas y comentarios en Internet que lo recuerdan. La implementación mejoró notablemente en Skylake, la sexta generación (2015). Yo tuve un Haswell durante muchos años, y empecé a probar AVX2 con esa máquina. Mi penúltimo portátil fue un Skylake. Ahora mismo tengo un PC con CPU de oncena generación, Tiger Lake, que es un pepino. Lo menciono porque, por casualidad, Tiger Lake es la última generación de procesadores de «consumo» que incluye las instrucciones AVX512: con la llegada de las CPU con núcleos heterogéneos, estas instrucciones se deshabilitan porque los núcleos «eficientes» no las implementan. Ahora mismo, si quieres AVX512, tienes que irte a Xeon, o a AMD.

Categorías
C#

Multiplicar por la traspuesta

Una operación muy frecuente con matrices consiste en multiplicar una matriz por la traspuesta de otra. La implementación más sencilla de esto sería trasponer la segunda matriz y luego multiplicar la primera matriz por la traspuesta. Lamentablemente, trasponer una matriz es poco eficiente. No hay un patrón sencillo de acceso a la caché que sea bueno al mismo tiempo: hay que descomponer la matriz en trozos pequeños… en pocas palabras, un lío. Y resulta que hay una forma más sencilla de conseguirlo: invertir la posición de los índices al acceder a la segunda matriz. El código «naïve» para la multiplicación directa de dos matrices se resumía en un triple bucle:

for (int i = 0; i < m; i++)
    for (int j = 0; j < p; j++)
    {
        double d = 0;
        for (int k = 0; k < n; k++)
            d += a[i, k] * b[k, j];
        result[i, j] = d;
    }

En la entrada sobre multiplicación de matrices, invertíamos el orden de los dos bucles interiores para mejorar el rendimiento de la caché de memoria. Además, añadíamos punteros e instrucciones AVX. Para la nueva rutina, lo que invertimos son los índices de la segunda matrix y, en consecuencia, renunciamos al truco de intercambiar los bucles internos. Perdemos algo de velocidad pero, en cualquier caso, el resultado es mejor que trasponer explícitamente y luego multiplicar.

La rutina que finalmente está incluida en Austra es la siguiente:

public unsafe Matrix MultiplyTranspose(Matrix m)
{
    Contract.Requires(IsInitialized);
    Contract.Requires(m.IsInitialized);
    if (Cols != m.Cols)
        throw new MatrixSizeException();
    Contract.Ensures(Contract.Result<Matrix>().Rows == Rows);
    Contract.Ensures(Contract.Result<Matrix>().Cols == m.Rows);

    int r = Rows;
    int n = Cols;
    int c = m.Rows;
    double[,] result = new double[r, c];
    fixed (double* pA = values, pB = m.values, pC = result)
    {
        int lastBlockIndex = n & CommonMatrix.AVX_MASK;
        double* pAi = pA;
        double* pCi = pC;
        for (int i = 0; i < r; i++)
        {
            double* pBj = pB;
            for (int j = 0; j < c; j++)
            {
                int k = 0;
                double acc = 0;
                if (Avx.IsSupported)
                {
                    Vector256<double> sum = Vector256<double>.Zero;
                    for (; k < lastBlockIndex; k += 4)
                        sum = Avx.Add(
                            left: Avx.Multiply(
                                left: Avx.LoadVector256(pAi + k),
                                right: Avx.LoadVector256(pBj + k)),
                            right: sum);
                    sum = Avx.HorizontalAdd(sum, sum);
                    acc = sum.ToScalar() + sum.GetElement(2);
                }
                for (; k < n; k++)
                    acc += pAi[k] * pBj[k];
                pCi[j] = acc;
                pBj += n;
            }
            pAi += n;
            pCi += c;
        }
    }
    return result;
}

Una parte importante del truco es hacer que el lenguaje que implementa Austra reconozca la multiplicación por la traspuesta y sepa usar la operación más eficiente. Pero eso es fácil de conseguir, y lo explicaré en otro momento.

Categorías
C#

Blazor

Empecé mis experimentos con Blazor en los lejanos tiempos de .NET 5. Hace un par de años, quiero decir.

Con un poco de contexto me explicaré mejor: prefiero una buena «aplicación de escritorio» antes que cualquier página web. Puede ser porque el tipo de proyectos en los que he invertido más tiempo han sido siempre aplicaciones con alta densidad de información, como las llaman ahora. Pero también porque he podido comparar la productividad de equipos trabajando para la web y para aplicaciones de escritorio de toda la vida. La pérdida de productividad, según mi experiencia personal, se agrava cuando es Angular la herramienta de front-end. Conozco TypeScript en profundidad, y he hecho proyectos con Angular y con Svelte. Svelte es muchísimo más productivo, e infinitamente menos pesado luego en producción. Pero la industria tiene cierta fijación con el puñetero Angular.

Las primeras pruebas con Blazor las hice con la idea de usarlo para crear prototipos rápidos. Funcionó de maravillas. Pero no estaba muy puesto en Bootstrap por entonces, y el diseño visual de los ejemplos de Microsoft era, y sigue siendo, abominable. Me preocupaba, además, la falta de controles de terceros. El cliente, por ejemplo, estaba empeñado en usar un gráfico de radar en la página principal del proyecto… y los componentes que manejábamos no traían el dichoso gráfico. Al final, el front-end terminó haciéndose en React, pero el prototipo de Blazor sigue existiendo y usándose internamente, porque en varios aspectos, es más rápido y potente.

Una de las lecciones más importantes de mi carrera como desarrollador, es que si te toca crear una librería, un servidor o algo que normalmente no tenga una interfaz de usuario… es mejor que te crees una interfaz propia, aunque no te la paguen ni te la agradezcan. Es tu seguro de vida para que no te culpen si el front-end o la aplicación hecha por otros es lenta. La vida es dura.

Ahora mismo, acabo de terminar un proyecto grande con Blazor y ASP.NET. Bueno, realmente el prototipo, porque los «sabios» de la «industria» siguen obcecados con Angular y Spring Boot, y ahora habrá que rehacerlo todo. Blazor es ya una herramienta de desarrollo madura. Yo he aprendido un poco más de CSS y HTML (desastres ambos, en mi opinión) y la parte de la fealdad ya está más o menos superada. Los componentes de terceros han evolucionado también. Y hay mejores libros y blogs sobre cómo funciona Blazor, que no siempre es lo que aparenta. La velocidad de desarrollo sigue siendo incomparable. He terminado la funcionalidad necesaria del proyecto en un mes. Calculamos que rehacerlo todo en Angular nos costará tres veces más tiempo. La estimación de tiempo no es mía: mi papel ha sido rebajarla, porque al fin y al cabo, ya sabemos cómo hacer todo.

Me siento un poco como Sísifo. Pero uno termina por acostumbrarse a estas cosas.

Categorías
C# FinTech

Cholesky

Esta va a ser una entrada «utilitaria»: poca explicación, pero aportando código fuente, por si alguien lo necesita. ¿Recuerda la entrada en la que presenté la descomposición de Cholesky? En aquel momento, no incluí el algoritmo de la descomposición, porque quería experimentar un poco con la implementación. Ya lo he hecho, y estoy más o menos satisfecho con el resultado.

Antes de ver el código, le explico el contexto. Este es un método de una estructura Matrix, que encapsula solamente una matriz bidimensional de valores flotantes (campo values). Dentro de la estructura se definen todos los operadores y métodos que podéis imaginar. En paralelo, defino otra estructura, LowerMatrix, para representar matrices triangulares inferiores. No hay relaciones de herencia, al tratarse de estructura, pero la clase de matrices triangulares permite definir métodos y operadores más eficientes. Es importante tener en cuenta que LowerMatrix gasta exactamente la misma memoria que Matrix. La ventaja está en esos métodos que evitan procesar las dos mitades de la matriz.

También hace falta saber que he definido operadores de conversión implícitos que transforman un array bidimensional en uno u otro tipo de matrices. Este operador es el que me permite evitar un constructor explícito para devolver un valor en el método:

public unsafe LowerMatrix Cholesky()
{
    int n = Rows;
    double[,] dest = new double[n, n];
    double[,] src = values;
    double* tmp = stackalloc double[n + n];

    // First column is special.
    double ajj = src[0, 0];
    if (ajj <= 0)
    {
        dest[0, 0] = double.NaN;
        return dest;
    }
    dest[0, 0] = ajj = Math.Sqrt(ajj);
    double r = 1 / ajj;
    n--;
    for (int i = 1; i <= n; i++)
        dest[i, 0] = src[i, 0] * r;
    for (int j = 1; j <= n; j++)
    {
        // Compute the diagonal cell.
        double v = 0.0;
        for (int i = 0; i < j; i++)
        {
            double a = dest[j, i];
            v += a * a;
        }
        ajj = src[j, j] - v;
        if (ajj <= 0)
        {
            dest[j, j] = double.NaN;
            return dest;
        }
        dest[j, j] = ajj = Math.Sqrt(ajj);

        // Compute the other cells of column J.
        if (j < n)
        {
            r = 1 / ajj;
            for (int i = 0; i < j; i++)
                tmp[i] = dest[j, i];
            for (int i = j; i < n; i++)
            {
                v = 0.0;
                for (int k = 0; k < j; k++)
                    v += dest[i + 1, k] * tmp[k];
                tmp[i] = v;
            }
            for (int i = j; i < n; i++)
                dest[i + 1, j] = (src[i + 1, j] - tmp[i]) * r;
        }
    }
    return dest;
}

Sobre el algoritmo, en sí: el algoritmo de Cholesky puede fallar cuando la matriz de origen no es positiva semidefinida. Mi implementación detecta ese caso al calcular las raíces cuadradas… y se limita a parar, poniendo un NaN en la celda diagonal donde se ha detectado el problema. Esto significa que el método, en la práctica, asume que la matriz es positiva semidefinida. En mi código, tengo un segundo método, TryCholesky, que devuelve un valor lógico para ver si la conversión fue posible, y retorna la matriz transformada como parámetro de salida.

Desde el punto de vista de un programador de C#, el único detalle interesante es el uso de stackalloc para reservar un array de memoria en la pila, en vez de usar memoria dinámica. Esto es lo que obliga a declarar el método con unsafe.

En rendimiento, el método es más rápido que la «versión base» de la librería que he visto que es más rápida usando sólo C# (usando Intel MKL, no hay competencia posible). Me refiero a la versión base porque, para matrices grandes, las librerías serias suelen dividir la matriz en bloques que se pueden procesar en paralelo. Este código, como ve, no usa threads, instrucciones SIMD y sólo utiliza punteros para la caché. En menos palabras: todo es mejorable.

Categorías
C#

Entran una matriz y un vector en un bar

… y claro, al rato sale un vector «transformado».

Esta entrada no es, aunque pueda parecerlo, un ripio de la anterior. Algorítmicamente, transformar un vector con una matriz se parece mucho a una sucesión de productos escalares. Pero resulta que el producto escalar, al menos hasta AVX2, tiene su truco. Vamos a comenzar por la implementación más tonta:

public static double[] Mult(double[,] a, double[] x)
{
    int m = a.GetLength(0);
    int n = a.GetLength(1);
    double[] b = new double[m];
    for (int = 0; i < m; i++)
    {
        double d = 0;
        for (int j = 0; j < n; j++)
            d += a[i, j] * x[j];
        b[i] = d;
    }
    return b;
}

Recordemos que tenemos un «handicap» autoimpuesto por representar las matrices como arrays bidimensionales de C#. Pero esta vez no voy a dar la brasa con los punteros, que ya sabemos que resuelven este problema sin pestañear. Esta es la implementación final que necesitamos, con soporte opcional de AVX para cuando esté disponible y merezca la pena:

public static unsafe double[] Mult(double[,] a, double[] x)
{
    int m = a.GetLength(0);
    int n = a.GetLength(1);
    double[] b = new double[m];
    int lastBlockIndex = n - (n % 4);
    fixed (double* pA = a)
    fixed (double* pX = x)
    fixed (double* pB = b)
    {
        double* pA1 = pA;
        double* pB1 = pB;
        if (n >= 12 && Avx2.IsSupported)
            for (int i = 0; i < m; i++)
            {
                int j = 0;
                var v = Vector256<double>.Zero;
                while (j < lastBlockIndex)
                {
                    v = Avx.Add(
                        v,
                        Avx.Multiply(
                            Avx.LoadVector256(pA1 + j),
                            Avx.LoadVector256(pX + j)));
                    j += 4;
                }
                v = Avx.HorizontalAdd(v, v);
                double d = v.ToScalar() + v.GetElement(2);
                for (; j < n; j++)
                    d += pA1[j] * pX[j];
                *pB1 = d;
                pA1 += n;
                pB1++;
            }
        else
            for (int i = 0; i < m; i++)
            {
                int j = 0;
                double d = 0;
                while (j < lastBlockIndex)
                {
                    d += (*(pA1 + j) * *(pX + j)) +
                        (*(pA1 + j + 1) * *(pX + j + 1)) +
                        (*(pA1 + j + 2) * *(pX + j + 2)) +
                        (*(pA1 + j + 3) * *(pX + j + 3));
                    j += 4;
                }
                for (; j < n; j++)
                     d += pA1[j] * pX[j];
                *pB1 = d;
                pA1 += n;
                pB1++;
            }
    }
    return b;
}

Esta vez, el código SIMD sólo se usa cuando hay doce o más elementos en el vector. La cifra la he elegido experimentando en mi i7-4770. Puede que en otros ordenadores, el umbral sea más bajo incluso.

Tengo que explicar cómo se implementa un producto escalar con SIMD, porque no es muy evidente. Uno diría que hay que acumular un escalar en una variable global al bucle… pero no hay ninguna operación SIMD que calcule directamente la suma de las cuatro multiplicaciones necesarias. La explicación oficial es que una suma de ese tipo destrozaría el paralelismo de la CPU. Y yo me lo creo, de veras. La consecuencia es que necesitamos acumular las multiplicaciones en cuatro variables; es decir, en un vector que hace de acumulador.

Las cosas se ponen de color hormiga cuando terminamos el bucle y tenemos entonces que sumar los cuatro elementos del vector acumulador. Analicemos las líneas 27 y 28 del listado anterior. Según mis experimentos, es la forma más rápida de conseguirlo. HorizontalAdd, cuando se trata de Vector256<double>, suma el primer elemento con el segundo, y lo almacena por partida doble en el primer y segundo elemento. A la vez, suma el tercero y el cuarto y hace lo mismo para guardar el resultado. Los métodos de extensión ToScalar() y GetElement() acceden entonces directamente al primer y tercer elemento y los suma. Mantengo la llamada inicial a HorizontalAdd porque, teóricamente, puede hacer dos de las sumas en paralelo, pero puedes experimentar a ver qué pasa si accedes directamente a los cuatro elementos y los sumas como toda la vida. A mí ya se me ha acabado la partida de tiempo libre para este experimento.

La razón para la controversia es que, en realidad, Internet está lleno de recomendaciones para hacer esta suma final de esta otra manera:

v = Avx2.Permute4x64(
    Avx.HorizontalAdd(v, v),
    0b00_10_01_11);
double d = Avx.HorizontalAdd(v, v).ToScalar();
// v = Avx.HorizontalAdd(v, v);
// double d = v.ToScalar() + v.GetElement(2);

Es decir: se llama dos veces a HorizontalAdd, pasando entre medias por una permutación entre escalares. En la arquitectura Haswell, al menos, esto funciona más lento que mi solución.

Si multiplico una matriz aleatoria de 64×64 por un vector de 64 elementos, obtengo estas cifras:

Method Mean Error StdDev Median
MultVector 5.762 μs 0.1142 μs 0.2227 μs 5.646 μs
FMultVector 1.814 μs 0.0320 μs 0.0416 μs 1.818 μs

No está mal, aunque no conseguimos tanta ventaja como con la multiplicación entre matrices. La versión con punteros y sin SIMD tampoco va mal, pero queda muy claro que el SIMD acelera este código. De paso, ya tenemos un patrón de código para productos escalares (y para cosas más raras como multiplicar un vector de sensibilidad delta-gamma por un escenario histórico: cosas de la valoración de productos financieros).

Por cierto, el mejor chiste que conozco sobre gente que entra en un bar tiene que ver con la Mecánica Cuántica. Dice así: entra el Gato de Schrödinger en un bar… y no entra.

Categorías
C#

Multiplicación de matrices

Supongamos que queremos multiplicar un par de matrices, $A$ y $B$. Digamos que la primera tiene dimensiones $m\times n$ y que la segunda es $n\times p$. La coincidencia entre columnas de la primera y filas de la segunda es condición necesaria para que podamos multiplicarlas.

Si me piden que escriba de carrerilla un método para esta multiplicación, esto es lo que se me ocurre:

public static double[,] Mult(double[,] a, double[,] b)
{
    int m = a.GetLength(0);
    int n = a.GetLength(1);
    int p = b.GetLength(1);
    double[,] result = new double[m, p];
    for (int i = 0; i < m; i++)
        for (int j = 0; j < p; j++)
        {
            double d = 0;
            for (int k = 0; k < n; k++)
                d += a[i, k] * b[k, j];
            result[i, j] = d;
        }
    return result;
}

He utilizado matrices bidimensionales de C# porque acceder a sus elementos individuales es sencillo. Internamente, C# las almacena en una sola memoria contigua de memoria, fila por fila.

El código que he mostrado no es una maravilla. Para empezar, cada vez que decimos algo como a[i, k], el compilador tiene que multiplicar la variable i por el número de columnas y por los ocho bytes que tiene un flotante de doble precisión. Hacerlo una vez no es problema… pero tenemos tres bucles anidados. Eso tiene que doler. Si en vez de C# escribiésemos esto en C++, el compilador podría sustituir un montón de multiplicaciones por sumas. RyuJIT ha mejorado muchísimo, pero no tanto.

C#, además, es un lenguaje mucho más seguro que C++, pero esta seguridad nos cuesta un montón de verificaciones de rango para poder indexar. Recordemos, además, que cada acceso necesita dos índices.

Y hay un tercer problema, mucho más sutil: cuando las matrices son grandes, el código anterior machaca la caché de la CPU sin piedad. Toma un folio de papel y haz el experimento: dibuja dos matrices, y ve numerando las celdas siguiendo el orden en que las usa el algoritmo.

La clase Unsafe

Llegados a este punto, tenemos dos alternativas: o marcamos el método como unsafe y usamos directamente punteros de C#, o intentamos evitarlo haciendo uso de la clase Unsafe, de System.Runtime.CompilerServices. Vamos a comenzar por esta última. De paso, voy a invertir el orden de los dos bucles más internos, para ver qué conseguimos con ello. Este es el código modificado, y suele funcionar el doble de rápido, o un poco más:

public static double[,] Mult(double[,] a, double[,] b)
{
    int m = a.GetLength(0);
    int n = a.GetLength(1);
    int p = b.GetLength(1);
    double[,] c= new double[m, p];
    ref double rA = ref a[0, 0];
    ref double rB = ref b[0, 0];
    ref double rC = ref c[0, 0];
    for (int i = 0; i < m; i++)
    {
        ref double rAi = ref Unsafe.Add(ref rA, i * n);
        ref double rCi = ref Unsafe.Add(ref rC, i * n);
        for (int k = 0; k < n; k++)
        {
            double d = Unsafe.Add(ref rAi, k);
            int kp = k * p;
            for (int j = 0; j < p; j++)
                Unsafe.Add(ref rCi, j) +=
                    d * Unsafe.Add(ref rB, kp + j);
        }
    }
    return c;
}

La regla principal del uso de Unsafe.Add es que si inicializamos así:

ref double rA = ref a[0, 0];

entonces el acceso a a[i, j] debe parecerse a esto:

Unsafe.Add(ref rA, i * n + j) = 42;

Esa multiplicación es un problema del que ya advertimos. En nuestro código lo paliamos moviendo la multiplicación al inicio del bucle donde se le da valor al índice de la fila. Mi apaño no es la palabra definitiva: le dejo como ejercicio la eliminación total de esas multiplicaciones.

Ahora hay que prestar atención, sobre todo, al patrón de acceso a memoria que se produce en el bucle más interno. En el algoritmo inicial, acumulábamos todos los términos de un elemento de la matriz final en el bucle interno, y asignábamos su suma de golpe a la celda del resultado. Esta variante, sin embargo, no parece tan buena. Tenemos que asumir que, al reservar memoria para la matriz, todas sus entradas valen cero (y es así). Luego, cada celda del resultado se va rellenando por pasos, no de una vez. Puede que esto sea bueno para la caché de la CPU, pero no me queda tan claro que sea bueno para el compilador de C#.

Pero lo que nos interesa realmente es que ahora ejecutamos el siguiente patrón de cálculo:

  1. Tenemos dos zonas de memoria consecutiva.
  2. Leemos algo de la primera zona.
  3. Lo transformamos como sea.
  4. Lo asignamos a la celda equivalente en la segunda zona de memoria.

Instrucciones SIMD

Ese patrón de actividad es el típico algoritmo «vectorial» que podemos acelerar utilizando operaciones SIMD. Tenemos dos opciones:

  • Utilizar System.Numerics.Vector, que se adapta automáticamente a cualquier máquina que soporte SIMD, e incluso ofrece una alternativa cuando no existe ese soporte. Este tipo funciona también para .NET Framework, a través de un paquete.
  • Si podemos usar .NET Core 3.1, podemos ir directamente a las clases declaradas en System.Runtime.Intrinsics y System.Runtime.Intrinsics.X86. Es un poco más complicado y no está bien documentado, pero da resultados ligeramente mejores.

Vamos a ir directamente por la segunda vía. Vamos a optimizar las CPUs que soporten el conjunto de instrucciones AVX, haremos algo más en el caso en que soporte el conjunto FMA (que mezcla multiplicaciones y sumas en una misma operación) y, de todas maneras, habilitaremos código de respaldo para cuando el procesador no soporte SIMD.

Cuando hay soporte para instrucciones AVX, podemos procesar hasta cuatro variables de tipo double de una tacada. Para ello tenemos que utilizar el tipo de estructura Vector256, que tiene capacidad para cuatro elementos. La forma más sencilla de inicializar estos vectores es utilizando punteros, por lo que vamos a tener que declarar nuestro método unsafe y pasarnos directamente a los punteros.

public static unsafe double[,] Mult(double[,] a, double[,] b)
{
    int m = a.GetLength(0);
    int n = a.GetLength(1);
    int p = b.GetLength(1);
    double[,] c = new double[m, p];
    int lastBlockIndex = p - (p % 4);
    fixed (double* pA = a)
    fixed (double* pB = b)
    fixed (double* pC = c)
    {
        double* pAi = pA;
        double* pCi = pC;
        for (int i = 0; i < m; i++)
        {
            double* pBk = pB;
            for (int k = 0; k < n; k++)
            {
                double d = *(pAi + k);
                if (Avx.IsSupported)
                {
                    int j = 0;
                    var vd = Vector256.Create(d);
                    while (j < lastBlockIndex)
                    {
                        if (Fma.IsSupported)
                            Avx.Store(pCi + j,
                                Fma.MultiplyAdd(
                                Avx.LoadVector256(pBk + j),
                                vd,
                                Avx.LoadVector256(pCi + j)));
                        else
                            Avx.Store(pCi + j,
                                Avx.Add(
                                Avx.LoadVector256(pCi + j),
                                Avx.Multiply(
                                Avx.LoadVector256(pBk + j),
                                vd)));
                        j += 4;
                    }
                    while (j < p)
                    {
                        pCi[j] += d * pBk[j];
                        j++;
                    }
                }
                else
                {
                    for (int j = 0; j < p; j++)
                        pCi[j] += d * pBk[j];
                }
                pBk += p;
            }
            pAi += n;
            pCi += p;
        }
    }
    return c;
}

Observaciones:

  1. Lo peor de trabajar con SIMD es tener que lidiar con vectores que no son múltiplos exactos del tamaño del vector básico. Nuestros vectores básicos tienen cuatro elementos. Si tenemos un vector de 75 elementos, necesitaremos un bucle de 18 repeticiones que procese cuatro elementos por vez, para una mierdecilla de bucle final que maneje los 3 elementos que nos sobran.
  2. Aunque la llamada a Avx.IsSupported está metida dentro de dos bucles anidados, no se preocupe: el compilador JIT la trata como una constante en tiempo de generación de código nativo, y no cuesta nada. Si no se soporta AVX, el compilador JIT solamente genera el código de la cláusula else, que funciona sobre cualquier arquitectura.
  3. Ojo: ese código «para cualquier máquina» podría optimizarse echando mano de la técnica de loop unrolling. Pero mi política en estos casos es: si no tienes una máquina decente, jódete.
  4. En el ejemplo anterior, cuando intercambiamos el orden de los bucles más internos, teníamos un valor escalar que sacábamos fuera del tercer bucle. Pero SIMD no ofrece instrucciones para multiplicar un vector por un escalar: tenemos que convertir ese escalar en todo un vector y utilizar la instrucción de multiplicación más general. No es grave, de todos modos.
  5. Si, además de AVX, la máquina soporta el conjunto FMA de instrucciones, podemos utilizar el método MultiplyAdd para acelerar un poco el algoritmo. Pero con esto hay que tener cuidado: a * b + c puede dar resultados diferentes si se hacen las dos operaciones por separado o a la vez. Si se hacen a la vez, aumenta la exactitud de la operación al existir menos redondeos. Pero el efecto secundario es que los cálculos con y sin esa opción dan resultados ligeramente diferentes. Tenemos que decidir cuándo es aceptable que exista esa diferencia y cuándo no. En cualquier caso, tengamos presente que el resultado de MultiplyAdd es más preciso.

Benchmark.NET

Para estar seguro de las ganancias en velocidad, he utilizado el package Benchmark.NET para generar las pruebas. Estos son los resultados:

Method Mean Error StdDev
MultMatrix 4,482.3 μs 88.75 μs 138.17 μs
UMultMatrix 1,895.2 μs 37.87 μs 63.26 μs
FMultMatrix 506.3 μs 3.44 μs 2.87 μs

La mejora por el uso de SIMD es cerca de cuatro veces, porque es el número de operaciones simultáneas que permite esta arquitectura en particular. Con AVX512 tendríamos vectores de ocho valores, pero necesitaríamos procesadores mucho más modernos, y de momento .NET Core no lo soporta.

Para esta prueba, he utilizado matrices de 128×128. He probado también con matrices de 8×8 e incluso de 4×4. La ganancia no es tan espectacular, pero en total se consigue una cuarta parte del tiempo de ejecución respecto al algoritmo más sencillo.

Categorías
C#

Un ejemplo de estructura

Para ver algo más de las nuevas posibilidades de las estructuras de C#, voy a mostrar una clase sencilla que utilizo en Iridium, un motor de valoración de swaps, en su versión nativa para .NET Core. La estructura (¡qué sorpresa!) sirve para representar fechas pero, a diferencia de DateTime, sin la parte de la hora.

Internamente, una fecha se representa como el número de días transcurridos desde el uno de enero del año 1. Como son fechas para software financiero, me la trae al viento los problemas de cambio de calendario. Con estas premisas, puedo representar el número de días con un valor entero de 32 bits, con lo que mi tipo Date ocupa la mitad del espacio que un DateTime. Esta es la declaración de la estructura y de su único campo de estado:

/// <summary>A date with efficient operations.</summary>
public readonly struct Date : IEquatable<Date>, IComparable<Date>
{
    /// <summary>Number of days since Jan 1st, 1.</summary>
    private readonly int date;
    public Date(int year, int month, int day) { ... }
}

Para ganar velocidad, además, el constructor no verifica que los componentes de una fecha sean correctos: en Iridium, eso es responsabilidad del generador de cupones y otras partes del código que generan fechas.

Ahora viene la parte más interesante del tipo de datos:

public void Deconstruct(
     out int year, out int month, out int day) { ... }

Un deconstructor es un método que permite, precisamente, extraer en una sola operación las partes integrantes de una instancia de un tipo. Se introdujeron pensando en las tuplas, pero en realidad se pueden usar con cualquier clase o estructura, y es una pena que DateTime no cuente con uno de estos métodos. Gracias al deconstructor, podemos ejecutar instrucciones como la siguiente:

var (y1, m1, day1) = fromDate;

Si tuviese que usar DateTime, tendría que llamar por separado a las tres propiedades Year, Month y Day de la fecha. ¿El problema? Pues que para recuperar el año a partir de la representación interna hay que ejecutar un pequeño algoritmo que lleva su tiempo. Si luego quiero el mes, no importa: tengo que volver a ejecutar la parte que extrae el año. Y lo mismo pasa al pedir el día del mes. Con el deconstructor, en cambio, sólo tengo que descomponer la fecha en partes una sola vez, y obtengo los tres componentes. De hecho, la implementación de mi propiedad Day se permite el lujo de usar internamente el deconstructor:

public int Day
{
    get
    {
        Deconstruct(out _, out _, out int d);
        return d;
    }
}

Los subrayados son comodines para descartar el año y el mes.

¿Más cosas que pueden interesar? Por ejemplo, se permiten las conversiones de tipo entre Date y DateTime, en ambos sentidos, y permito convertir un valor Date en un número (la conversión inversa se consigue más elegantemente con un constructor adicional):

public static explicit operator DateTime(Date d) =>
    new DateTime(d.date * TicksPerDay);
public static explicit operator Date(DateTime d) =>
    new Date((int)(d.Ticks / TicksPerDay));
public static explicit operator int(Date d) => d.date;

El código completo de la clase puede descargar desde este enlace.

Categorías
C#

Estructuras en C#

Hay un antes y un después en C# en lo que atañe a los tipos struct, y tiene que ver con la llegada de RyuJIT, más que con .NET Core. El compilador de código nativo ha aprendido a manejar eficientemente estos tipos de datos.

Todos sabemos que las estructuras se diferencian de las clases en que son tipos con semántica de asignación por valor, en vez de la semántica habitual de asignación por referencia. Una variable de tipo struct contiene directamente los datos, mientras que una variable de clase es realmente un puntero a la zona de memoria donde residen los datos.

Lo quiero explicar en esta entrada, sin embargo, es cuándo es eficiente y aconsejable utilizar un tipo de estructura en vez de un tipo de clase. Hay un par de casos, y el primero de ellos es el más evidente: cuando el espacio que ocupan los campos de la estructura es suficientemente pequeño como caber en un registro de hardware.

Por ejemplo, .NET representa las fechas mediante el tipo de estructura DateTime. Este tipo guarda internamente un campo de tipo long, que representa el número de tics transcurridos desde cierta fecha base. Esto cabe en un registro de 64 bits. Para que todo vaya sobre ruedas, además, la estructura se declara de sólo lectura:

public readonly struct DateTime { }

En realidad, .NET Core es quien declara el tipo de sólo lectura. .NET Framework no lo hace, pero trata el tipo como si lo fuese. El cualificador readonly es relativamente nuevo en C#.

Para entender la utilidad de este caso de uso (disfrazar un tipo de datos numérico) hay que comparar con lo que ocurriría en un lenguaje como Java, que de momento no permite definir tipos de valor. O metes el entero en una clase, y cada vez que necesitas una fecha necesitas pedir memoria, o defines un montón de métodos que trabajen directamente con un entero… y te buscas la vida luego para averiguar cuándo un entero es un número de verdad o está representando una fecha.

Por supuesto, hay otra posibilidad: permitir modificaciones sobre las instancias de tu clase Java. Pero así es menos elegante y te arriesgas a todo tipo de errores. Y, de todas maneras, si necesitas crear 10.000 fechas, tienes otros tantos objetos ocupando memoria dinámica.

El segundo caso

Es más complicado explicar el segundo caso, por lo que voy a recurrir a un ejemplo en uno de mis proyectos: necesitaba un tipo para representar matrices. Las operaciones sobre ellas no formaban parte de la ruta crítica para la velocidad del código, pero la transformación de un vector por una matriz sí era crítica. Además, iba a necesitar muchas instancias de matrices.

Mi solución fue declarar la matriz como una estructura de sólo lectura. No se trata de un tipo pequeño, pues cada una necesita nueve números flotantes de doble precisión. Pero de esta manera ahorré memoria y, muy importante, una indirección innecesaria cada vez que tenía que transformar un vector.

El hacer que el tipo sea de sólo lectura es importante en estos casos, sobre todo si quiere declarar operadores. Mi multiplicación de matrices, por ejemplo, se implementa mediante un operador:

public static Matrix operator *(in Matrix m1, in Matrix m2){}

Si no utilizo los modificadores in en los parámetros, las matrices se pasan por copia. Pero si utilizo in y el tipo no es de sólo lectura, me sigo arriesgando a que el compilador haga una copia preventiva de la matriz, porque al fin y al cabo, in sólo significa que el operador no va a modificar el parámetro en su implementación. Es un poco raro, pero es lo que hay.

Hay que tener en cuenta, además, que en un método normal puedo pasar parámetros mediante ref, pero esto no es posible para un operador.

Un caso más

En mi código, por motivos históricos, hay un tercer caso intermedio: el de los vectores. Mis vectores no son de sólo lectura: quizás sería recomendable, pero me da pereza rescribir todo el código y hacer pruebas para comprobar que todo va mejor. Sin embargo, hago uso profuso del modificador in con vectores pasados como parámetros. ¿Cómo evito el desastre de la copia preventiva? He aquí la respuesta:

public readonly Vector Scale(in Vector factor) =>
    new Vector(X * factor.X, Y * factor.Y, Z * factor.Z);

C# permite declarar métodos con el modificador readonly, para garantizar que el método no modifica campos de la estructura. De este modo, si un método que recibe un vector como parámetro in ejecuta el método Scale u otro método de sólo lectura, el compilador evita crear una copia defensiva del vector.

Más novedades

Hay muchas más novedades en las últimas versiones de C# que tienen que ver con las estructuras (ref struct y ref readonly, por ejemplo), pero en esta entrada me he limitado a tratar algunas de las características que ayudan a la mejor generación de código nativo.