Най-бързият начин за четене и писане на двоичен код

В момента оптимизирам приложение, една от операциите, които се правят много често, е четене и запис на двоичен файл. Имам нужда от 2 вида функции:

Set(byte[] target, int index, int value);

int Get(byte[] source, int index);

Тези функции са необходими за знакови и неподписани short, int и long в голям и малък ред.

Ето няколко примера, които направих, но имам нужда от оценка на предимствата и недостатъците:

първият метод използва Marshal, за да запише стойността в паметта на byte[], вторият използва обикновени указатели, за да постигне това, а третият използва BitConverter и BlockCopy, за да направи това

unsafe void Set(byte[] target, int index, int value)
{
    fixed (byte* p = &target[0])
    {
        Marshal.WriteInt32(new IntPtr(p), index, value);
    }
}

unsafe void Set(byte[] target, int index, int value)
{
    int* p = &value;
    for (int i = 0; i < 4; i++)
    {
        target[offset + i] = *((byte*)p + i);
    }
}

void Set(byte[] target, int index, int value)
{
    byte[] data = BitConverter.GetBytes(value);
    Buffer.BlockCopy(data, 0, target, index, data.Length);
}

А ето и методите Read/Get:

първият използва Marshal за четене на стойността от byte[], вторият използва обикновени указатели, а третият отново използва BitConverter:

unsafe int Get(byte[] source, int index)
{
    fixed (byte* p = &source[0])
    {
        return Marshal.ReadInt32(new IntPtr(p), index);
    }
}

unsafe int Get(byte[] source, int index)
{
    fixed (byte* p = &source[0])
    {
        return *(int*)(p + index);
    }
}

unsafe int Get(byte[] source, int index)
{
    return BitConverter.ToInt32(source, index);
}

проверката на границите трябва да се направи, но все още не е част от въпроса ми...

Ще се радвам, ако някой може да каже кой би бил най-добрият и най-бърз начин в този случай или да ми даде други решения, върху които да работя. За предпочитане е генерично решение


Току-що направих някои тестове за ефективност, ето резултатите:

Set Marshal: 45 ms, Set Pointer: 48 ms, Set BitConverter: 71 ms Get Marshal: 45 ms, Get Pointer: 26 ms, Get BitConverter: 30 ms

изглежда, че използването на указатели е бързият начин, но мисля, че Marshal и BitConverter правят някои вътрешни проверки... може ли някой да потвърди това?


person haze4real    schedule 10.01.2010    source източник
comment
Имате кода, защо не го стартирате и не тествате с Stopwatch?   -  person mmx    schedule 10.01.2010
comment
:/ :\ прав си, ще направя това за бързо и ще редактирам въпроса си, но това не е единствената точка на публикацията ми. Търся алтернативи и може би общи начини да направя това също   -  person haze4real    schedule 10.01.2010
comment
Повдигнати вежди при този въпрос: преобразуването в двоичен трябва да е необходимо само за I/O. Самата I/O операция винаги е с няколко порядъка по-бавна от масажирането на битовете. Най-добрата оптимизация не може да ви купи повече от няколко процента подобрение.   -  person Hans Passant    schedule 10.01.2010


Отговори (4)


Важно: ако имате нужда само от one endian, вижте магията на указателя от wj32 / dtb


Лично аз бих писал директно в Stream (може би с известно буфериране) и бих използвал повторно споделен буфер, който като цяло мога да приема, че е чист. След това можете да направите някои преки пътища и да приемете индекс 0/1/2/3.

Със сигурност не използвайте BitConverter, тъй като това не може да се използва както за little/big-endian, което ви е необходимо. Също така бих бил склонен просто да използвам bit-shifting, а не unsafe и т.н. Всъщност е най-бързият въз основа на следното (така че се радвам, че това е начинът, по който вече го правя моя код тук, потърсете EncodeInt32Fixed):

Set1: 371ms
Set2: 171ms
Set3: 993ms
Set4: 91ms <==== bit-shifting ;-p

код:

using System;
using System.Diagnostics;
using System.Runtime.InteropServices;
static class Program
{
    static void Main()
    {
        const int LOOP = 10000000, INDEX = 100, VALUE = 512;
        byte[] buffer = new byte[1024];
        Stopwatch watch;

        watch = Stopwatch.StartNew();
        for (int i = 0; i < LOOP; i++)
        {
            Set1(buffer, INDEX, VALUE);
        }
        watch.Stop();
        Console.WriteLine("Set1: " + watch.ElapsedMilliseconds + "ms");

        watch = Stopwatch.StartNew();
        for (int i = 0; i < LOOP; i++)
        {
            Set2(buffer, INDEX, VALUE);
        }
        watch.Stop();
        Console.WriteLine("Set2: " + watch.ElapsedMilliseconds + "ms");

        watch = Stopwatch.StartNew();
        for (int i = 0; i < LOOP; i++)
        {
            Set3(buffer, INDEX, VALUE);
        }
        watch.Stop();
        Console.WriteLine("Set3: " + watch.ElapsedMilliseconds + "ms");

        watch = Stopwatch.StartNew();
        for (int i = 0; i < LOOP; i++)
        {
            Set4(buffer, INDEX, VALUE);
        }
        watch.Stop();
        Console.WriteLine("Set4: " + watch.ElapsedMilliseconds + "ms");

        Console.WriteLine("done");
        Console.ReadLine();
    }
    unsafe static void Set1(byte[] target, int index, int value)
    {
        fixed (byte* p = &target[0])
        {
            Marshal.WriteInt32(new IntPtr(p), index, value);
        }
    }

    unsafe static void Set2(byte[] target, int index, int value)
    {
        int* p = &value;
        for (int i = 0; i < 4; i++)
        {
            target[index + i] = *((byte*)p + i);
        }
    }

    static void Set3(byte[] target, int index, int value)
    {
        byte[] data = BitConverter.GetBytes(value);
        Buffer.BlockCopy(data, 0, target, index, data.Length);
    }
    static void Set4(byte[] target, int index, int value)
    {
        target[index++] = (byte)value;
        target[index++] = (byte)(value >> 8);
        target[index++] = (byte)(value >> 16);
        target[index] = (byte)(value >> 24);
    }
}
person Marc Gravell    schedule 10.01.2010
comment
Не мисля, че Stream би било добро решение, проблемът е, че може да са необходими търсения и данните не винаги се четат и записват последователно. Друг проблем би бил endianity.. - person haze4real; 10.01.2010
comment
първо трябва да потвърдя това и може да приема този отговор като решение, какво ще кажете за получаване/четене? - person haze4real; 10.01.2010
comment
Същото, но наобратно; return ((int)buffer[index++]) | (((int)buffer[index++]) << 8) | (((int)buffer[index++]) << 16) | (((int)buffer[index]) << 24); (или превключете преместванията отгоре надолу, за да получите друг край). Имайте предвид, че прехвърляме към int рано, тъй като int аритметиката е по-бърза от байт аритметиката. - person Marc Gravell; 10.01.2010
comment
Мисля, че решението за преместване е най-доброто в момента, голямото предимство би било лекотата на размяната на endian. - person haze4real; 10.01.2010

Използвайки Set1 до Set4 на Marc Gravell и Set5 по-долу, получавам следните числа на моята машина:

Set1: 197ms
Set2: 102ms
Set3: 604ms
Set4: 68ms
Set5: 55ms <==== pointer magic ;-p

Код:

unsafe static void Set5(byte[] target, int index, int value)
{
    fixed (byte* p = &target[index])
    {
        *((int*)p) = value;                
    }
}

Разбира се, става много по-бързо, когато масивът от байтове не е фиксиран на всяка итерация, а само веднъж:

Set6: 10ms (little endian)
Set7: 85ms (big endian)

Код:

if (!BitConverter.IsLittleEndian)
{
    throw new NotSupportedException();
}

watch = Stopwatch.StartNew();
fixed (byte* p = buffer)
{
    for (int i = 0; i < LOOP; i++)
    {
        *((int*)(p + INDEX)) = VALUE;
    }
}
watch.Stop();
Console.WriteLine("Set6: " + watch.ElapsedMilliseconds + "ms");

watch = Stopwatch.StartNew();
fixed (byte* p = buffer)
{
    for (int i = 0; i < LOOP; i++)
    {
        *((int*)(p + INDEX)) = System.Net.IPAddress.HostToNetworkOrder(VALUE);
    }
}
watch.Stop();
Console.WriteLine("Set7: " + watch.ElapsedMilliseconds + "ms");
person dtb    schedule 10.01.2010
comment
проблемът тук би бил режийните разходи, генерирани от endian swaping - person haze4real; 10.01.2010
comment
Готино; Добавих актуализация, но endianness все още е проблем тук. Днес вече нямам гласове, но виртуален +1 - person Marc Gravell; 10.01.2010
comment
@haze4real: режийните разходи, причинени от размяната на endian, всъщност не са толкова големи. Примерът е актуализиран. - person dtb; 10.01.2010
comment
:/ режийните разходи са огромни, около 8 пъти по-бавни във вашия тест. Не трябва да закачате масива от байтове извън цикъла във вашия тест, защото това е функцията, която трябва да бъде тествана... не мога да го закача между тези извиквания на функция - person haze4real; 10.01.2010
comment
Вярно, но Set7 все още е само с няколко милисекунди по-бавен от Set4. Моята препоръка би била решението на Marc за изместване на битове. - person dtb; 10.01.2010
comment
Да, но само защото сте фиксирали масива извън цикъла, което просто симулира количество извиквания на функции и не е част от решението, защото самата функция се извиква на различни места в кода. - person haze4real; 10.01.2010
comment
Точно така, можете да закачите масива извън цикъла само ако искате да промените същия масив няколко пъти подред. Ако това не е така във вашето приложение, не можете да направите това. Но ако можете, печалбата в производителността е очевидна. - person dtb; 10.01.2010

Указателите са пътят. Закрепването на обекти с ключовата дума fixed е изключително евтино и избягвате излишните разходи за извикване на функции като WriteInt32 и BlockCopy. За „общо решение“ можете просто да използвате void* и да използвате свой собствен memcpy (тъй като имате работа с малки количества данни). Указателите обаче не работят с истински генерични продукти.

person wj32    schedule 10.01.2010
comment
Тогава явно не сте написали кода си правилно. Наистина ли мислите, че използването на bit-shifting (около 8 инструкции за всеки Int32 най-малко ) е по-бързо от използването на проста инструкция за mov? И аз говоря за фиксиране на буфера извън цикъла. - person wj32; 10.01.2010
comment
можете ли да ми дадете пример за това в c#? говориш за втория метод Set, нали? просто имам нужда от типовете обикновени стойности: Int16, UInt16, Int32, UInt32, Int64, UInt64 - person haze4real; 10.01.2010
comment
фиксиран (байт* b = масив) { за (...) (int)(b + отместване) = стойност; } }. Точно това, което имате във вашия метод Get. Погледнете прозореца за разглобяване, ако не вярвате, че това всъщност е най-бързият метод. - person wj32; 10.01.2010
comment
А, правилно; Имах грешния край на пръчката. Но изискването е за двете крайни. Така че ще трябва да имате резервен вариант за другия и някакъв механизъм за абстрахиране на двете. Бих го запазил просто и бих написал смените. Това също така повдига въпроса Правя ли си труда да проверявам/поддържам хардуер с голям край и т.н. - person Marc Gravell; 10.01.2010
comment
Не би трябвало обаче да се нуждае от for цикъл (от коментар)... освен ако не съм пропуснал нещо? Просто вземете b + отместването, разменете на int* и присвоете? - person Marc Gravell; 10.01.2010
comment
За съжаление C# не предлага начин за използване на инструкцията bswap за съвместимост с голям/малък ред, така че в този случай вашето решение ще бъде най-бързото. Що се отнася до цикъла for, той трябваше да покаже, че закрепването ще се извърши извън всякакви цикли. - person wj32; 10.01.2010
comment
Готино; Добавих актуализация, но (както забелязвате) endianness все още е проблемът тук. Днес вече нямам гласове, но виртуален +1 - person Marc Gravell; 10.01.2010
comment
да, big endian е необходим много пъти, повечето от устройствата, с които комуникираме, използват big endian ред. - person haze4real; 10.01.2010

Трябва да направите известно профилиране на вашия код, за да разкриете дали това е тясното място. Освен това, гледайки кода ви, изглежда, че използвате извиквания на функция .Net, за да запишете един байт в неуправляван масив, включвайки пин на паметта и извикване на опасен код...

Може да е много по-добре да декларирате .Net System.IO.MemoryStream и да търсите и пишете около него, където е възможно да използвате записващ поток, за да накарате вашите промени, което трябва да използва по-малко извиквания на функции и няма да изисква опасен код. Ще откриете, че нещата с указателя са много по-полезни в C#, ако правите неща като DSP, където трябва да извършите една операция към всяка стойност в масив и т.н.

РЕДАКТИРАНЕ: Позволете ми също да спомена, че в зависимост от това, което правите, може да откриете, че кеширането на процесора ще влезе в сила, ако можете да продължите да работите върху една малка област от паметта, която се побира в кеша, тогава ще получите най-доброто производителност.

person Spence    schedule 10.01.2010
comment
Проблемът е, че може да бъде пречка, защото приложението комуникира с множество различни мрежови устройства и работи на евтини машини, някои от тези устройства използват тежки протоколи, други не. Знаете ли добър начин за профилиране на интерфейсите? проблемът ще бъде променливото забавяне на мрежата.. - person haze4real; 10.01.2010