Реализация reference counting или жизнь без GC (почти)

Доброго времени суток, хабр!

Многие считают, что системный язык и сборщик мусора — не совместимые понятия. В некоторых ситуациях, действительно, сборщик может доставлять некоторые проблемы.

Как Вам, скорее всего, известно — в D сборщик мусора, отчасти, опционален. Но ручное управление памятью это прошлый век.
Поэтому сегодня я покажу как можно реализовать сборку мусора самому через «полуавтоматический» подсчёт ссылок, а так же как при этом минимизировать обращения к встроенному в runtime сборщика мусора на основе сканирования памяти.



Решать мы будем задачу подсчёта ссылок на указатели, классы и массивы.
Сразу следует обговорить 2 момента: почему от GC в этой статье мы полностью не будем отказываться и почему подсчёт ссылок «полуавтоматический».
Полный отказ от GC подразумевает использование атрибута @nogc для всего кода, но тут есть одно НО. Интерфейс IAllocator, который мы будем использовать, позволяет создавать и уничтожать экземпляры класса правильно одной командой (правильно это с вызовом всех конструкторов и все деструкторов иерархии классов), но функции, которые это делают не помечены как @nogc и чтобы не раздувать статью мы не будем их реализовывать самостоятельно (отчасти в прошлой статье это было рассмотренно).
Подсчёт ссылок будет «полуавтоматическим», так как на данном этапе развития языка нельзя выполнить автоматически какой-то код при передаче или присвоении ссылочних типов (классы и массивы), поэтому этот код мы будет вызывать сами, но постараемся сделать это максимально скрыто.

Начнём с реализации allocator'а.

static this() { RC.affixObj = allocatorObject(RC.affix); } // оборачиваем в объект

struct RC
{
static:
    alias affix = AffixAllocator!(Mallocator,size_t,size_t).instance; // об этом ниже
    IAllocator affixObj;

    // сразу при создании инкрементируем счётчик
    auto make(T,A...)( auto ref A args )
    { return incRef( affixObj.make!T(args) ); }

    // напрямую мы не можем высвободить объект, только через уменьшение счётчика
    private void dispose(T)( T* p ) { affixObj.dispose(p); }

    private void dispose(T)( T p )
        if( is(T == class) || is(T == interface) )
    { affixObj.dispose(p); }

    // affix.prefix( void[] p ), где p -- область выделенной памяти,
    // а не просто указатель на неё, поэтому выглядит так некрасиво
    ref size_t refCount(T)( T p )
        if( is(T == class) || is(T == interface) )
    { return affix.prefix( (cast(void*)p)[0..__traits(classInstanceSize,T)] ); }

    ref size_t refCount(T)( T* p )
    { return affix.prefix( (cast(void*)p)[0..T.sizeof] ); }

    // увеличение счётчика, возвращаем объект для удобства
    auto incRef(T)( auto ref T p )
    {
        if( p is null ) return null;
        refCount(p)++;
        return p;
    }

    // уменьшение счётчика, возвращаем объект для удобства, null если счётчик равен 0
    auto decRef(T)( T p )
    {
        if( p is null ) return null;
        if( refCount(p) > 0 )
        {
            refCount(p)--;
            if( refCount(p) == 0 )
            {
                dispose(p); 
                return null;
            }
        }
        return p;
    }
}

В основе нашего аллокатора лежит Mallocator — аллокатор, работающий через C-шные malloc и free. Мы его обернули в AffixAllocator. Он параметризируется используемым настоящим аллокатором и 2-мя типами данных. При выделении памяти AffixAllocator выделяеть чуть больше: разме�� Prefix и Suffix типов (соответственно второй и третий параметр шаблонизации). Как можно догадаться префикс находится до выделенной под объект памяти, а суффикс после. В нашем случае Prefix и Suffix это size_t, и как раз префикс у нас и будет олицетворять счётчик ссылок.
Такой подход позволяет без модификации использовать уже существующие классы, так как информация о количестве ссылок хранится вне объекта.

Теперь мы можем создавать и уничтожать объекты классов и указатели с помощью нашего аллокатора.

    auto p = RC.make!int( 10 );
    assert( is( typeof(p) == int* ) );
    assert( *p == 10 );
    assert( RC.refCount(p) == 1 );
    p = RC.decRef(p);
    assert( p is null );

Уже что-то, но пока не интересно: только make за нас инкрементирует счётчик, далее инкрементируем и декриментируем мы его самостоятельно.

Добавим обёртку, которая будет некоторые вещи делать за нас.

struct RCObject(T)
{
    T obj;
    alias obj this; // где будет нужно, компилятор просто подставит obj поле вместо самого объекта RCObject

    this(this) { incRef(); } // конструктор копирования -- увеличиваем счётчик

    this( T o ) { obj = o; incRef(); } // создаётся новая обёртка -- увеличиваем счётчик

    // должна быть возможность поместить объект в обёртку без изменения счётчика ссылок (когда он приходит сразу из RC.make)
    package static auto make( T o )
    {
        RCObject!T ret;
        ret.obj = o;
        return ret;
    }

    // nothrow потому что этого требует деинциализатор из std.experimental.allocator
    ~this() nothrow
    {
        if( obj is null ) return;
        try  decRef();
        catch(Exception e)
        {
            import std.experimental.logger;
            try errorf( "ERROR: ", e );
            catch(Exception e) {}
        }
    }

    void incRef()
    {
        if( obj is null ) return;
        RC.incRef(obj);
    }

    /// удалит obj если кроме этой ссылки больше нет ни одной
    void decRef()
    { 
        if( obj is null ) return;
        assert( refCount > 0, "not null object have 0 refs" );
        obj = RC.decRef(obj);
    }

    size_t refCount() @property const
    {
        if( obj is null ) return 0;
        return RC.refCount(obj);
    }

    // при присвоении для старого объекта уменьшается счётчик ссылок, а после присвоения нового прибавляется
    void opAssign(X=this)( auto ref RCObject!T r )
    {
        decRef();
        obj = r.obj;
        incRef();
    }

    /// тоже самое только работа с голым типом T, без обёртки
    void opAssign(X=this)( auto ref T r )
    {
        decRef();
        obj = r;
        incRef();
    }
}

// для удобства
auto rcMake(T,A...)( A args ){ return RCObject!(T).make( RC.make!T(args) ); }

Теперь у нас есть scope-зависимый подсчёт ссылок и мы можем делать так:

    static class Ch  { }

    {
        RCObject!Ch c;
        {
            auto a = rcMake!Ch();
            assert( a.refCount == 1 );
            auto b = a;
            assert( a.refCount == 2 );
            assert( b.refCount == 2 );
            c = a;
            assert( a.refCount == 3 );
            assert( b.refCount == 3 );
            assert( c.refCount == 3 );
            b = rcMake!Ch();
            assert( a.refCount == 2 );
            assert( b.refCount == 1 );
            assert( c.refCount == 2 );
            b = rcMake!Ch(); // тут старый объект удалится, а новый запишется
            assert( a.refCount == 2 );
            assert( b.refCount == 1 );
            assert( c.refCount == 2 );
        }
        assert( c.refCount == 1 );
    }

Это уже что-то! Но как же быть с массивами? Добавим работу с массивами в наш аллокатор:

...
    T[] makeArray(T,A...)( size_t length )
    { return incRef( affixObj.makeArray!T(length) ); }

    T[] makeArray(T,A...)( size_t length, auto ref T init )
    { return incRef( affixObj.makeArray!T(length,init) ); }

    private void dispose(T)( T[] arr ) { affixObj.dispose(arr); }

    ref size_t refCount(T)( T[] arr ) { return affix.prefix( cast(void[])arr ); }
...

И реализуем обёртку для массивов.

она мало отличается от обёртки для объектов

struct RCArray(T)
{
    // считаем ссылки на память, которую выделили
    private T[] orig;
    // а работать можем со срезом
    T[] work;

    private void init( T[] origin, T[] slice )
    {
        if( slice !is null )
            assert( slice.ptr >= origin.ptr &&
                    slice.ptr < origin.ptr + origin.length,
                    "slice is not in original" );

        orig = origin;
        incRef();

        work = slice is null ? orig : slice;

        static if( isRCType!T ) // если элементы являются обёртками
            foreach( ref w; work ) w.incRef; // добавляем счётчик только рабочему набору
    }

    ///
    alias work this;

    this(this) { incRef(); } конструктор копирования

    this( T[] orig, T[] slice=null ) { init( orig, slice ); }

    /// no increment ref
    package static auto make( T[] o )
    {
        RCArray!T ret;
        ret.orig = o;
        ret.work = o;
        return ret;
    }

    // срез возвращает обёртку
    auto opSlice( size_t i, size_t j )
    { return RCArray!T( orig, work[i..j] ); }

    void opAssign(X=this)( auto ref RCArray!T arr )
    {
        decRef();
        init( arr.orig, arr.work );
    }

    void incRef()
    {
        if( orig is null ) return;
        RC.incRef(orig);
    }

    void decRef()
    {
        if( orig is null ) return;
        assert( refCount > 0, "not null object have 0 refs" );
        orig = RC.decRef(orig);
    }

    size_t refCount() @property const
    {
        if( orig is null ) return 0;
        return RC.refCount(orig);
    }

    ~this()
    {
        if( refCount )
        {
            // логический хак
            if( refCount == 1 )
            {
                static if( isRCType!T )
                    foreach( ref w; orig )
                        if( w.refCount )
                            w.incRef;
            }

            static if( isRCType!T ) // если элементы обёртки
                foreach( ref w; work ) w.decRef;
            decRef;
        }
    }
}

template isRCType(T)
{
    static if( is( T E == RCObject!X, X ) || is( T E == RCArray!X, X ) )
        enum isRCType = true;
    else
        enum isRCType = false;
}

но есть некоторые принципиальные моменты:

1. в обёртке для массивов мы храним массив выделенной памяти и рабочий срез
2. в конструкторе, если элементы являются обёртками, увеличиваем для элементов рабочего среза счётчик ссылок
3. в деструкторе в этой же ситуации уменьшаем

Так же в деструкторе есть небольшой логический хак: допустим у нас сохранился только срез массива, а обёртка на оригинальный массив канула в лету. Тогда получается, что счётчик ссылок у orig равняется 1, это значит, что если серз будет тоже удалён, то будет вызван dispose для изначального (orig) массива, это приведёт к тому, что объекты, взятые из оригинального массива, но не попадающие в срез будут подвергнуты операции уменьшения счётчика ссылок и могут быть удалены. Чтобы это не произошло мы добавляем +1 каждому элементу, у которого уже есть больше 1. Потом будет происходить уменьшение у рабочего набора, это позволит оставить на 1 больше у элементов оригинального массива, которые не вошли в рабочий набор и при удалии оригинального массива их счётчик не дойдёт до нуля.

Всё это вместе позволяет делать нам вот такие вещи:

    class A
    {
        int no;
        this( int i ) { no=i; }
    }

    alias RCA = RCObject!A;

    {
        RCA obj;
        {
            RCArray!RCA tmp1;
            {
                RCArray!RCA tmp2;
                {
                    auto arr = rcMakeArray!RCA(6);
                    foreach( int i, ref a; arr )
                        a = rcMake!A(i);
                    assert( arr[0].refCount == 1 );
                    assert( arr[1].refCount == 1 );
                    assert( arr[2].refCount == 1 );
                    assert( arr[3].refCount == 1 );
                    assert( arr[4].refCount == 1 );
                    assert( arr[5].refCount == 1 );

                    tmp1 = arr[1..4];
                    assert( arr[0].refCount == 1 );
                    assert( arr[1].refCount == 2 );
                    assert( arr[2].refCount == 2 );
                    assert( arr[3].refCount == 2 );
                    assert( arr[4].refCount == 1 );
                    assert( arr[5].refCount == 1 );

                    tmp2 = arr[3..5];
                    assert( arr[0].refCount == 1 );
                    assert( arr[1].refCount == 2 );
                    assert( arr[2].refCount == 2 );
                    assert( arr[3].refCount == 3 );
                    assert( arr[4].refCount == 2 );
                    assert( arr[5].refCount == 1 );

                    obj = tmp2[0];
                    assert( arr[0].refCount == 1 );
                    assert( arr[1].refCount == 2 );
                    assert( arr[2].refCount == 2 );
                    assert( arr[3].refCount == 4 );
                    assert( arr[4].refCount == 2 );
                    assert( arr[5].refCount == 1 );
                }
                assert( tmp1[0].refCount == 1 );
                assert( tmp1[1].refCount == 1 );
                assert( tmp1[2].refCount == 3 );

                assert( obj.refCount == 3 );

                assert( tmp2[0].refCount == 3 );
                assert( tmp2[0].obj.no == 3 );
                assert( tmp2[1].refCount == 1 );
                assert( tmp2[1].obj.no == 4 );
            }
            assert( tmp1[0].refCount == 1 );
            assert( tmp1[1].refCount == 1 );
            assert( tmp1[2].refCount == 2 );
            assert( obj.refCount == 2 );
        }
        assert( obj.refCount == 1 );
    }
    // тут будет удалён последний элемент с индексом 3

Хоть код и не помечен как @nogc, он не обращается к встроенному GC. А как мы знаем не трогай и не запахнет не выделяй через GC и он не будет собирать.

На этом всё, надеюсь Вы что-то новое для себя подчерпнули)

Код оформлен как пакет dub.
Сорцы на github.

Это не готовая библиотека, а пока просто набросок, он требует ещё много доработок.
Буду очень рад помощи, если не делом, так советом.