System Design

На собеседованиях для разработчиков часто есть секция про системный дизайн (System Design). Как правило на ней просят спроектировать какую-нибудь более или менее сложную систему. Например сервис аналогичный tinyurl, или систему полнотекстового поиска, или еще что-то в этом духе. Чем выше позиция, на которую претендуешь, тем важнее/сложнее эта секция. Меня пару раз просили дать список статей, которые позволят подготовиться к этому. Далее некоторая подборка, разбитая по темам.

Как делать деплой:
How Twitter deploys its widgets JavaScript
Manhattan software deployments: how we deploy Twitter’s large scale distributed database
Создание системы непрерывного развертывания: опыт Instagram

Как делать хранилище данных (правильнее сказать persistent layer):
The Google File System
Bigtable: A Distributed Storage System for Structured Data

Spanner: Google’s Globally-Distributed Database

The Log: What every software engineer should know about real-time data's unifying abstraction

Amazon dynamoDB: a seamlessly scalable non-relational database service

Building DistributedLog: Twitter’s high-performance replicated log service

Управление ресурсами кластера:

Borg, Omega, and Kubernetes

Балансировка нагрузки:

The tail at scale

Обработка данных:

Flying faster with Twitter Heron

MapReduce: Simplified Data Processing on Large Clusters 

Spark RDD

Best practices:

Software Engineering Advice from Building Large-Scale Distributed Systems

Five Considerations For Large Scale Systems

Консенсус:

Consensus in the wild

"Building Scalable Stateful Services" by Caitie McCaffrey

GOTO 2015 • Applying the Saga Pattern • Caitie McCaffrey

Опыт компаний в построении систем:

A 360 Degree View Of The Entire Netflix Stack
A Brief History of Scaling LinkedIn
Redundant Array of Inexpensive Datacenters

Также есть тематические блоги, на которых публикуется много интересной и полезной информации. Ниже список тех, на которые я подписан:
LinedIn Engineering
Metadata
AMP Lab blog
Aphyr
All Things Distributed
Badoo Tech
Clouder Engineering Blog
Dropbox Engineering Blog
Twitter Engineering Blog
Facebook Code
Google Research Blog
Highscalability
Indeed Engineering
NGINX Blog
Netflix Thech Blog

Tail latency

Tail latency - это когда, например, по 90-му процентилю сервер обрабатывает наши запросы за 50 мллисекунд, а по 99-му за 1 секунду. Это очень плохо, т.к. некоторые пользователи будут ждать ответа непростительно долго.
В этой статье рассматриваются некоторые возможные источники tail latency
В таблице ниже сводка из статьи:

Специальные функции-члены в C++ 2

Здесь я рассказывал про специальные функции члены, при каких условиях они генерируются, при каких нет. При этом я не указал важный нюанс, касающийся шаблонов.
Допустим у нас есть код:

struct MyStruct {
    MyStruct() = default;

    template <typename T>
    MyStruct &operator=(const T&) {
        std::cout << "template copy assignment\n";
        return *this;
    }

    template <typename T>
    MyStruct(const T&) {
        std::cout << "template copy ctor\n";
    }
};

int main() {
    MyStruct c1;
    MyStruct c2 = c1;
    c1 = c2;
}

На экран, как не странно, не будет выведено ничего. В предыдущем посте я писал о том, что если определить оператор копирующего присваивания или конструктор копирования, то компилятор не генерирует оператор копирующего/перемещающего присваивания и конструктор копирования/перемещения. Но подобное правило не работает для шаблонных функций, поэтому компилятор сгенерировал все версии специальных функций по-умолчанию.

Про инициативность от Eric Niebler

Best Professional Advice I Ever Got

Как устроен weak_ptr

Для чего нужен.

В предыдущем посте я рассказывал про устройство shared_ptr. К сожалению, он подвержен классической проблеме циклических ссылок. Для решения этой проблемы используется weak_ptr.

Как решить проблему циклических ссылок.

В объект счётчика ссылок добавляется счётчик слабых ссылок. weak_ptr увеличивает этот счётчик при создании и копировании. При этом объект счётчика ссылок не удаляется до тех пор, пока есть хотя бы один shared_ptr или один weak_ptr, т.к. weak_ptr необходимо обращаться к счётчику ссылок даже если нет ни одного живого shared_ptr. Но сам объект, на который указывает shared_ptr, может быть удалён, если больше не осталось живых shared_ptr.

Код объекта счётчика ссылок теперь будет выглядеть так:

template <typename T>
class sp_counted {
public:
    explicit sp_counted(T *p) noexcept
        : shared_count(1),
          weak_count(1),
          ptr(p) {}

    void add_ref() noexcept {
        ++shared_count;
    }

    void release() noexcept {
        if (!--shared_count) {
            // Если последний shared_ptr удалился, удаляем объект
            delete ptr;
            if (!--weak_count) {
                // Если нет слабых ссылок, удаляем объект счётчика
                destroy();
            }
        }
    }

    void add_ref_weak() noexcept {
        ++weak_count;
    }

    void release_weak() noexcept {
        if (!--weak_count) {
            // Если последний weak_ptr удалился, удаляем объект счётчика.
            // Т.к. shared_ptr тоже увеличивает weak_count при создании,
            // нет необходимости проверять значение shared_count
            destroy();
        }
    }

    size_t use_count() const noexcept {
        return shared_count.load();
    }

    // Попытка увеличить счётчик shared_count из weak_ptr
    // Потокобезопасен, lock-free
    void add_ref_lock() {
        // Сохраняем текущее значение shared_count
        size_t cur_value(shared_count.load());
        do {
            // Если счётчик сильных ссылок равен нулю (т.е. нет больше живых shared_ptr),
            // то новый shared_ptr создавать не из чего.
            if (!cur_value) {
                throw std::bad_weak_ptr();
            }
          // Пытаемся увеличить счётчик shared_count на единицу
          // Если в промежутке между сохранением shared_count в cur_value, shared_count изменился,
          // то операция compare_exchange_weak вернёт false, запишет новое значение shared_count в cur_value,
          // и цикл повторится
        } while (shared_count.compare_exchange_weak(cur_value, cur_value + 1));
    }

private:
    void destroy() noexcept {
        delete this;
    }

private:
    // Счётчик ссылок shared_ptr
    std::atomic<size_t> shared_count;
    // Счётчик ссылок weak_ptr
    std::atomic<size_t> weak_count;
    T *ptr;
};

Вопрос: почему вызов add_ref_lock безопасен? Ведь в другом потоке shared_ptr в деструкторе может удалить объект sp_counted, и обращение к переменной shared_count будет некорректным.
Ответ: потому что деструктор shared_ptr не удалит объект sp_counted, т.к. есть живая слабая ссылка (weak_count != 0).
Код weak_ptr будет выглядеть так:

template <typename T>
class weak_ptr {
    friend class shared_ptr<T>;
public:
    weak_ptr() noexcept : ptr(nullptr), counted(nullptr) {}

    weak_ptr(const weak_ptr &other) noexcept : ptr(other.ptr), counted(other.counted) {
        add_ref_weak();
    }

    weak_ptr(const shared_ptr<T> &p) noexcept : ptr(p.ptr), counted(p.counted) {
        add_ref_weak();
    }

    weak_ptr &operator=(const weak_ptr &other) noexcept {
        release_weak();

        ptr = other.ptr;
        counted = other.counted;

        add_ref_weak();

        return *this;
    }

    weak_ptr &operator=(const shared_ptr<T> &p) {
        release_weak();

        ptr = p.ptr;
        counted = p.counted;

        add_ref_weak();

        return *this;
    }

    // Пытаемся сделать shared_ptr. Для этого вызывается конструктор shared_ptr(const weak_ptr &amp;);
    // В случае невозможности создать shared_ptr возвращается пустой объект
    shared_ptr<T> lock() noexcept {
        try {
            return shared_ptr<T>(*this);
        } catch (const std::bad_weak_ptr &) {
            return shared_ptr<T>();
        }
    }

    size_t use_count() const noexcept {
        return counted != nullptr ? counted->use_count() : 0;
    }

private:
    void add_ref_weak() noexcept {
        if (counted) {
            counted->add_ref_weak();
        }
    }

    void release_weak() noexcept {
        if (counted) {
            counted->release_weak();
        }
    }

private:
    T *ptr;
    sp_counted<T> *counted;
};

В код shared_ptr добавляется конструктор shared_ptr(const weak_ptr &p):

template <typename T>
class weak_ptr;

template <typename T>
class shared_ptr {
    friend class weak_ptr<T>;
public:
    shared_ptr() noexcept : ptr(nullptr), counted(nullptr) {}

    // excaption safe конструктор
    explicit shared_ptr(T *p) {
        std::unique_ptr<T> holder(p);
        // new может кинуть исключение, и если p не передать в unique_ptr,
        // память под p потеряется
        counted = new sp_counted<T>(holder.get());
        ptr = holder.release();
    }

    ~shared_ptr() noexcept {
        release();
    }

    shared_ptr(const shared_ptr<T> &other) noexcept : ptr(other.ptr), counted(other.counted) {
        add_ref();
    }

    shared_ptr(const weak_ptr<T> &p) : ptr(p.ptr), counted(p.counted) {
        if (counted) {
            // Пытаемся увеличить счётчик ссылок объекта
            // В случае неудачи сгенерируется исключение std::bad_weak_ptr()
            counted->add_ref_lock();
        } else {
            throw std::bad_weak_ptr();
        }
    }

    shared_ptr &operator=(const shared_ptr<T> &other) noexcept {
        // Освобождаем владение предыдущим указателем
        release();

        // Выполняем присваивание
        ptr = other.ptr;
        counted = other.counted;

        // Устанавливаем владение новым указателем
        add_ref();

        // Ура! Я не забыл вернуть *this!
        return *this;
    }

    T *get() const noexcept {
        return ptr;
    }

    size_t use_count() const noexcept {
        return counted != nullptr ? counted->use_count() : 0;
    }

private:
    void add_ref() noexcept {
        if (counted) {
            counted->add_ref();
        }
    }

    void release() noexcept {
        if (counted) {
            counted->release();
        }
    }

private:
    T *ptr;
    sp_counted<T> *counted;
};

А что с потокобезопасностью?

Гарантии потокобезопасности такие же как и у shared_ptr. Но на всякий случай повторю: любая операция с weak_ptr потокобезопасна, если каждый поток хранит копию weak_ptr. При этом каждая копия может хранить сслыку на один shared_ptr. Хранить ссылки на один weak_ptr в разных потоках непотокобезопасно.

Как устроен shared_ptr

Для чего нужен

shared_ptr нужен для ситуации, когда имеются несколько ссылок на объект, созданных в динамической памяти. При этом освобождать объект нужно тогда, когда удаляется последняя ссылка. И делать это нужно безопасно даже в многопоточном коде.

Как отслеживать ссылки на объект

Каждый объект shared_ptr содержит созданный в динамической памяти счётчик ссылок, который увеличивается на единицу, когда shared_ptr завладевает объектом, и уменьшается на единицу, когда shared_ptr перестаёт владеть объектом.

Код объекта счётчика ссылок может выглядеть так:

template <typename T>
class sp_counted {
public:
    explicit sp_counted(T *p) noexcept : count(1), ptr(p) {}

    void add_ref() noexcept {
        ++count;
    }

    void release() noexcept {
        if (!--count) {
            delete ptr;
            delete this;
        }
    }

    size_t use_count() const noexcept {
        return count.load();
    }

private:
    std::atomic<size_t> count;
    T *ptr;
};

При этом код shared_ptr может выглядеть так:

template <typename T>
class shared_ptr {
public:
    shared_ptr() noexcept : ptr(nullptr), counted(nullptr) {}

    // excaption safe конструктор
    explicit shared_ptr(T *p) {
        std::unique_ptr<T> holder(p);
        // new может кинуть исключение, и, если p не передать в unique_ptr,
        // память под p потеряется
        counted = new sp_counted<T>(holder.get());
        ptr = holder.release();
    }

    ~shared_ptr() noexcept {
        release();
    }

    shared_ptr(const shared_ptr &other) noexcept : ptr(other.ptr), counted(other.counted) {
        add_ref();
    }

    shared_ptr &operator=(const shared_ptr &other) noexcept {
        // Освобождаем владение предыдущим указателем
        release();

        // Выполняем присваивание
        ptr = other.ptr;
        counted = other.counted;

        // Устанавливаем владение новым указателем
        add_ref();

        // Ура! Я не забыл вернуть *this!
        return *this;
    }

    T *get() const noexcept {
        return ptr;
    }

    size_t use_count() const noexcept {
        return counted != nullptr ? counted->use_count() : 0;
    }

private:
    void add_ref() {
        if (counted) {
            counted->add_ref();
        }
    }

    void release() {
        if (counted) {
            counted->release();
        }
    }

private:
    T *ptr;
    sp_counted<T> *counted;
};

А что с потокобезопасностью?

Любая операция с shared_ptr потокобезопасна, если каждый поток хранит копию shared_ptr. При этом каждая копия может хранить указатель на один объект.

Хранить ссылки на один shared_ptr в разных потоках непотокобезопасно.

Разработка модулей для nginx

Отличное руководство по разработке модулей для nginx - здесь

Алгоритмы кэширования

На досуге решил поразбираться с алгоритмами кеширования, и узнал что кроме всем известных LFU и LRU есть целый зоопарк различных алгоритмов со своими достоинствами и недостатками. Некоторые из них я изучил более подробно, написал свои реализации и сравнил эффективность. В качестве данных использовался лог запросов в базу данных картинок одного высоконагруженного сайта. Запросы не обладают свойством локальности, но обладают свойством частотности. То есть запрошенный элемент, как правило, запрашивается только по прошествию какого-то времени. Поэтому для такого потока данных лучше всего подходят алгоритмы кэширования обладающие ствойством LFU кэша. Это и будет видно на графиках.

OPT

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

FIFO

Если искомый элемент не находится в кеше, он вставляется в хвост очереди. Если нужно освободить место, удаляются элементы из головы очереди. Таким образом вытесняется элемент, находящийся в кэше дольше всех.
Реализация.

LFU

Каждый элемент имеет счётчик обращений. Новый элемент вставляется в кэш со значением счётчика равным 1. При попадании в кэш счётчик найденного элемента увеличивается на 1. Если нужно освободить место, нужно найти элемент с самым маленьким значением счётчика. Таким образом вытесняется тот элемент, которые запрашивался реже всего.
Главный недостаток LFU состоит в том, что некогда частотные элементы могут присутствовать в кэше очень долго, даже если они уже давно не запрашивались.
Реализация.

LRU

Новый элемент вставляется в голову списка. При попадании в кэш элемент перемещается в голову списка. Если нужно освободить место, вытесняется элемент из хвоста списка. Таким образом вытесняется тот элемент, которые не запрашивался дольше всех.
Главный недостаток LRU состоит в том, что он не устойчив к линейному чтению, т.к. оно вымывает все "прогретые" данные.
Реализация.

Сегментный LRU (SNLRU)

Один из недостатков LRU в том, что если мы вставили новый элемент в кэш, и этот элемент больше не будет никогда запрошен, ему придётся пройти через весь список, пока он не будет вытеснен. Сегментный LRU решает эту проблему. Кэш организован ввиде нескольких (N) LRU кэшей. Новый элемент вставляется в нулевой LRU кэш. При попадании в кэш элемент перемещается в следующий LRU кэш, либо на MRU (Most Recently Used) позицию последнего LRU кэша, если выше уже идти некуда. При вытеснении элемента из k-го LRU кэша он перемещается в k-1 LRU кэш. По достижению нулевого LRU кэша элемент удаляется.
Алгоритм устойчив к линейному чтению, т.к. новые элементы не могут вытеснить прогретые. При этом он обладает свойством LFU алгоритма - элементы перемещается в следующий список при каждом попадании в кэш. Более того, если взять N достаточно большим, то получится LFU кэш. При этом сегментный LRU обладает тем же недостатком LFU - частотные в прошлом элементы будут долго сидеть в кэше прежде чем вымоются оттуда.
Реализация.

Mid point LRU

Является вариацией сегментного LRU, в котором количество сегментов - 2, и эти два сегмента делят кэш не пополам, а в некоторой пропорции. Размеры сегментов подбирается эмпирически и зависит от характера потока данных. Этот алгоритм очень просто реализуется и на деле показывает один из лучших результатов. Обладает всеми достоинствами сегментного LRU.
Реализация.

2Q

Кэш разделяется на три части:

• In - FIFO кэш, в который попадают все новые элементы;
• Out - FIFO кэш, в который попадают элементы, вытесненные из In. При этом этот кэш хранит ключ и не хранит значение, поэтому его можно сделать достаточно большим;
• Main - главный LRU кэш, в который попадают новые элементы найденные в Out. При вытеснении элемента из главного кэша он удаляется.

Алгоритм устойчивость к линейному чтению.

Оригинальная статья с описанием, реализация.

MQ

По сути является сегментным LRU но с некоторыми существенными улучшениями.

Для каждого элемента ведётся счётчик его запросов. Номер LRU кэша, в который стоит поместить элемент вычисляется используя некоторую функцию от этого счётчика, например логарифм.

Возможна ситуация, когда элемент, который вытесняется из кэша, в ближайшем будущем может стать очень популярным, поэтому имеет смысл как-то запоминать позицию, с которой он был вытеснен, и при следующем добавлении в кэш этого элемента сразу ставить его на позицию с учётом его "прошлых заслуг". Для этого по аналогии с 2Q алгоритмом заводится FIFO кэш, который хранит ключ элемента и позицию, с которой элемент был вытеснен. Если новый добавляемый элемент находится в этом FIFO кэше, он сразу добавляется в позицию, с который был вытеснен в прошлом.

Алгоритм обладает также свойсвами LFU. Чтобы решить проблему элементов, которые часто запрашивались в прошлом, а сейчас просто занимают место, каждый элемент хранит время, когда к нему было обращение. Далее в какие-то моменты времени проверяются элементы на LRU позиции всех сегментов, и если к этим элементам давно не было оращения, они перемещаются в сегмент нижнего уровня, либо удаляются из кэша если находятся на нулевом сегменте.

Оригинальная статья с описанием, реализация.

Графики:

В легенде алгоритмы отсортированы от лучшего к худшему.
MQ* - MQ алгоритм, в котором не выполняется проверка времени жизни объекта в кэше.
На графике видно, что все существующие алгоритмы можно ещё оптимизировать, т.к. до теоретически оптимального есть ещё около 10%.

Специальные функции-члены в C++

Здесь есть хорошее описание специальных функции-членов С++ от Howard Hinnant (один из основных комиттеров в Clang STL). Специальные функции-члены - это те, которые компилятор генерирует, если они не объявлены явно:

• Конструктор по умолчанию (Default constructor);
• Деструктор (Destructor);
• Конструктор копирования (Copy constructor);
• Оператор копирующего присваивания (Copy assignment);
• Конструктор переноса (Move constructor);
• Оператор переносящего присваивания (Move assignment);

Или в коде:

struct MyStruct {
    MyStruct() = default;
    ~MyStruct() = default;
    MyStruct(const MyStruct &) = default;
    MyStruct &operator=(const MyStruct &) = default;
    MyStruct(MyStruct &&) = default;
    MyStruct &operator=(MyStruct &&) = default;
};

Каждая из этих функций может быть:

• Не объявлена;
• Может быть сгенерирована компилятором:
• Как default - вызывает соответствующие функции для членов. Для конструктора - конструкторы членов, для оператора присваивания - операторы присваивания членов и т.д.
• Как delete - если соответствующая функция у члена объявлена как delete или функция не может быть сгенерирована. Например для оператора присваивания член может иметь оператор присваивания объявленный как delete, или объект не может копироваться, т.к. содержит ссылку.
• Может быть объвлена явно программистом:
• Как default;
• Как delete;
• С телом функции;

Разница между "не объявлена" и объявлена с delete в том, что не объявленная функция не участвует в разрешении перегрузки.

Например класс:

struct MyStruct {
    template <typename... Args>
    MyStruct(Args...) {
    }
};

может быть создан вызовом конструктора без параметров, т.к. конструктор по умолчанию не объявлен, и не участвует в разрешении перегрузки.
Но такой класс:

struct MyStruct {
    template <typename... Args>
    MyStruct(Args...) {
    }
    MyStruct() = delete;
};

создать вызовом конструктора без параметров не получится, т.к. конструктор по умолчанию объявлен и участвует в разрешении перегрузки, и при этом он объявлен как delete.
При этом стоить помнить одно правило - если перемещающие функции объявленны компиляторм как delete, они не участвуют в разрешении перегрузки, не смотря на факт объвления.
Когда и какие функции компилятор генерирует сам, а когда объявляет их как delete?
Объявленный конструктор по умолчанию - остальные 5 функций объявляются компилятором как default:

struct MyStruct {
    MyStruct() = default;
    ~MyStruct() = default;
    MyStruct(const MyStruct &) = default;
    MyStruct &operator=(const MyStruct &) = default;
    MyStruct(MyStruct &&) = default;
    MyStruct &operator=(MyStruct &&) = default;
};

Объявленный деструктор - конструктор перемещения и перемещающий оператор присваивания становятся не объявленными, остальные функции объявляются компилятором как default:

struct MyStruct {
    MyStruct() = default;
    ~MyStruct() = default;
    MyStruct(const MyStruct &) = default;
    MyStruct &operator=(const MyStruct &) = default;


};

Объявленный конструктор копирования - конструктор по умолчанию, конструктор перемещения и перемещающий оператор присваивания становятся не объявленными, остальные функции объявляются компилятором как default:

struct MyStruct {

    ~MyStruct() = default;
    MyStruct(const MyStruct &) = default;
    MyStruct &operator=(const MyStruct &) = default;


};

Объявленный копирующий оператор присваивания - конструктор перемещения и перемещающий оператор присваивания становятся не объявленными, остальные функции объявляются компилятором как default:

struct MyStruct {
    MyStruct() = default;
    ~MyStruct() = default;
    MyStruct(const MyStruct &) = default;
    MyStruct &operator=(const MyStruct &) = default;


};

Объявленный конструктор перемещения - конструктор по умолчанию и перемещающий оператор присваивания становятся не объявленным, деструктор объявляется компилятором как default, остальные функции объявляются компилятором как delete.

struct MyStruct {

    ~MyStruct() = default;
    MyStruct(const MyStruct &) = delete;
    MyStruct &operator=(const MyStruct &) = delete;
    MyStruct(MyStruct &&) = default;

};

Объявленный перемещающий оператор присваивания - конструктор по умолчанию и деструктор объявляется компилятором как default, конструктор копирования и оператор копирующего присваивания объявляются компилятором как delete, конструктор перемещения становится не объявленным.

struct MyStruct {
    MyStruct() = default;
    ~MyStruct() = default;
    MyStruct(const MyStruct &) = delete;
    MyStruct &operator=(const MyStruct &) = delete;

    MyStruct &operator=(MyStruct &&) = default;
};

Презентация Facebook на ВМК МГУ

28 октября в на ВМК МГУ ребята из Facebook рассказывали про устройство Facebook TAO - хранилище социального графа.
Небольшая выжимка:

• Социальный граф - набор объектов и связи между ними. Например: пост связан с пользователем, фотография к посту связана с постом, коментарии и лайки других пользователей связаны с постом, пользователи могут иметь отношение дружбы, и т.д.
• Чтобы отрисовать страницу пользователя, нужно загрузить все ноды графа, связанные с последним событием;
• На загрузку одной ноды графа отводится 1 милисекунда;
• Особенность кеширования - наиболее горячие данные те, которые недавно создались;
• У каждого объекта графа есть уникальный 64 битный глобальный идентификатор. Уникальность гарантируется так: в идентификаторе объекта N бит отводится под идентификатор пользователя, остальные биты под событие. Главное для каждого пользователя уметь получить следующий уникальный идентификатор события. Это не сложно и не требует глобальной синхронизации.
• В качестве хранилища свой отбранчёванный MySQL с InnoDB;
• Многоуровневое кеширование c Memcached и McRouter;
• Есть процедура прогревания кеша;
• 99.8% - чтение данных, 0.2% - запись;
• 3 датацентра в США, 1 в Европе;
• В каждом ДЦ есть мастер нода MySQL. Все записи идут в мастер ноду, далее асинхронно реплицируются;
• Если мастер нода падает, выбирается новая мастер нода;
• Чтобы не задолбать бэкенд, над каждым инстансом MySQL располагается Master Cache, через который идут все запросы в этот инстанс. Это позволяет контролировать нагрузку на  MySQL;
• Очень большой кеш-хит, поэтому до бэкенда MySQL долетает мало запросов (около 1 KRPS на инстанс);
• Если все кеши отвалятся, MySQL бекенд нагрузку не выдержит;
• Для проведения аналитики по социальному графу есть выгрузка данных в Hive;
• Шардируют базу по диапазонам идентификатора, поэтому процедуры перешардирования нет;
• Есть региональные прокси для снижения латентности https handshake;

Более подробно можно почитать в статье

Материалы по модели памяти C++

1. C++ and Beyond 2012: Herb Sutter - atomic<> Weapons Part 1
2. C++ and Beyond 2012: Herb Sutter - atomic<> Weapons Part 2
3. C++ Concurrency in Action: Practical Multithreading (Chapter 5)
4. Pershing on programming
5. Threads Cannot be Implemented as a Library
6. Threads and memory model for C++ by Hans Boehm

C++ Type Deduction

На конференции cppcon 2014 был доклад Скота Мейерса про выведение типов - Type Deduction and Why You Care (слайды).
Небольшая шпаргалка:

#include <iostream>

template <typename T>
T func1(T f) {
    return f;
}

template <typename T>
T func2(T& f) {
    return f;
}

template <typename T>
T func3(const T& f) {
    return f;
}

template <typename T>
T func4(T&& f) {
    return f;
}

void func5(const int * const param1,
           const int *       param2,
                 int *       param3) {
    auto p1 = param1;   // p1 = const int*; same as for auto *
    auto p2 = param2;   // p2 = const int*; same as for auto *
    auto p3 = param3;   // p2 = int*; same as for auto *

    auto &rp1 = param1; // p1 = const int *const &; same as for auto &&
    auto &rp2 = param2; // p2 = const int *&; same as for auto &&
    auto &rp3 = param3; // p3 = int *&; same as for auto &&
}

int main(int argc, char* argv[]) {
    int x = 10;
    int &rx = x;
    const int &crx = x;

    func1(x);       // T = int; param = int;
    func1(rx);      // T = int; param = int;
    func1(crx);     // T = int; param = int;

    func2(x);       // T = int; param = int&;
    func2(rx);      // T = int; param = int&;
    func2(crx);     // T = const int; param = const int&;

    func3(x);       // T = int; param = const int&;
    func3(rx);      // T = int; param = const int&;
    func3(crx);     // T = int; param = const int&;

    func4(x);       // T = int&; param = int&;
    func4(rx);      // T = int&; param = int&;
    func4(crx);     // T = const int&; param = const int&;
    func4(42);      // T = int; param = int&&;

    return 0;
}

auto в range base loop

На всех конференциях посвящённых C++ практически всегда рекомендуют использовать для определения переменных auto. Один из доводов в пользу auto, кроме того, что приходится меньше писать, в том, что сгенерированный код быстрее, т.к. всегда выводится корректный тип, и нет места приведению типов. Хотя кажется что тип всегда и так очевиден возможны ситуации, когда по невнимательности типы не совпадают, и компилятор выполняет конвертацию.
Например:

#include <map>
#include <string>

int main() {
  std::map<std::string, int> m;
  for(const auto &a : m) {}
  for(const std::pair<std::string, int> &a : m) {}
  return 0;
}

Вопрос: Какой вариант for быстрее? Ответ: Первый, т.к. auto& выводит корректный тип элемента, который для std::map<std::string, int> - std::pair<const std::string, int>, т.к. для std::map запрещено модифицировать ключ. Для второго for-а создаётся временный объект std::pair<std::string, int>, поэтому он работает медленнее.

Грабли шаблонного конструктора

Допустим у нас есть шаблонный класс с шаблонным конструктором копий и функция create:

#include <string>

template <typename T>
class MyClass {
public:
    MyClass() = default;

    MyClass(T t) : value(t) {}

    template <typename U>
    MyClass(const MyClass<U>& other) : value(other.value) {}

    T value;
};

template <typename T>
MyClass<T> create(T v) {
    return MyClass<T>(v);
}

Также имеются две перегруженные функции:

void func(const MyClass<std::string> &) {}
void func(const MyClass<float> &) {}

Если попытаться скомпилировать такой код:

int main() {
    func(create("Hello"));
    return 0;
}

То получим "странное" сообщение об ошибке:

clang++ -Wall -Wextra -std=c++11 test.cpp -O3 -o test
test.cpp:28:5: error: call to 'func' is ambiguous
    func(create("Hello"));
    ^~~~
test.cpp:16:6: note: candidate function
void func(const MyClass &) {
     ^
test.cpp:19:6: note: candidate function
void func(const MyClass &) {
     ^
1 error generated.
make: *** [all] Error 1

То есть компилятор считает обе функции подходящими вариантами и не знает какую выбрать. На самом деле понятно почему. У MyClass есть шаблонный конструктор копий, который может принимать любой тип. Найдя его при разрешении перегрузки функции func, компилятор даже не проверяет, можно ли реально вызвать этот конструктор, и не будет ли ошибок. А ошибки будут, т.к. const char* не приводится к float.
Такая же проблема была с std::pair до принятие стандарта C++11. То есть такой код:

#include <utility>

void func(const pair<int, int>&) {}
void func(const pair<std::string, std::string>&) {}

int main(int argc, char* argv[]) {
    func(std::make_pair("Hello", "world"));

    return 0;
}

Приводил к такой же ошибке.
Решить эту проблему можно с помощью SFINAE и enable_if:

template <typename T>
class MyClass {
public:
    MyClass() = default;

    MyClass(T t) : value(t) {}

    template <typename U, typename = typename std::enable_if<std::is_convertible<U, T>::value>::type>
    MyClass(const MyClass<U>& other) : value(other.value) {}

    T value;
};

Теперь шаблонный конструктор копий участвует в перегрузке только в том случае, если тип T класса, который передаётся в конструктор конвертируется в тип T класса, который конструируется.
Если шаблонному классу std::enable_if в качестве первого параметра передали true, то внутри него определён typedef с типом второго парамтера. Если же первый параметр false, то typedef не определён.
Поле value в шаблонном классе std::is_convertible равно true или false в зависимости от того, конвертируется ли первый параметр шаблона во второй.
Итого, если в конструкции

std::enable_if<std::is_convertible<U, T>::value>::type>

тип U конвертируется в тип T, то std::enable_if::type определён, и функция учавствует в разрешении перегрузки. В противном случае std::enable_if::type не определён, и вся конструкция является некорректной. Такая ситуация называется "substitution failure". По стандарту она не является ошибкой компиляции (is not an error), и компилятор просто выкидывает шаблонную функцию из рассмотрения. Данный приём называется SFINAE - substitution failure is not an error.

HTML кодировщик для C++ кода

Ссылка

Проверка существоания методов у класса

На cppcon 2014 был доклад для про современное метопрограммирование шаблонов - слайды.

Там в частности рассказывалось как реализовать проверку существует ли для класса оператор копирующего присваивания. Суть метода в том, чтобы получить тип результата некоторого выражения (в данном случае оператора присваивания) через decltype, и инстанцировать этим типом шаблон. Если выражение корректно (то есть у нас есть оператор присваивания), то результатом параметризации будет один тип (например std::true_type), если нет, то другой тип (например std::false_type).

Пример кода из слайдов:

template< class T >
using copy_assign_t = decltype(declval<T&>() = declval< T const& >());

template<class T>
struct is_copy_assignable {
private:
  template< class U, class = copy_assign_t<U> >
  static true_type try_assign( U&& ); // SFINAE may apply!
  static false_type try_assign(...); // catch-all overload
public:
  using type = decltype(try_assign(declval<T>()));
};

Работает это всё благодаря SFINAE.

Я попробовал написать общий класс, который возвращает true_type/false_type в зависимости от того, является ли любое переданное выражение корректным. Его можно использовать, чтобы реализовать свои проверки для типов, при этом не требуется писать много кода.

Например нужно проверить, есть ли в классе нестатическая функция get_value():

//Объявляем тип выражения
template <typename T>
using has_get_value_t = decltype(std::declval<T>().get_value());

template <typename T>
struct has_get_value : is_correct_expression_t<T, has_get_value_t> { };

int main() {
  std::cout << std::is_same<std::true_type, has_get_value<MyClass>::type>::value << std::endl;
}

Как это реализовано:

template <typename T, template<class> class E>
struct is_correct_expression_t {
private:
    template <typename U, typename = E<U>>
    static std::true_type try_evaluate(U&&);
    static std::false_type try_evaluate(...);
public:
    using type = decltype(try_evaluate(std::declval<T>()));
};

Для move/copy assign будет выглядеть так:

template <typename T>
using copy_assign_t = decltype(std::declval<T&>() = std::declval<const T&>());

template <typename T>
struct is_copy_assignable : is_correct_expression_t<T, copy_assign_t> { };

template <typename T>
using move_assign_t = decltype(std::declval<T&>() = std::declval<T&&>());

template <typename T>
struct is_move_assignable : is_correct_expression_t<T, move_assign_t> { };

Трюк со стековой памятью

Интересный способ оптимизации я недавно обнаружил изучая реализацию системного вызова poll в ядре Linux. Этому системному вызову передаётся массив структур pollfd. Для работы с этим массивом его необходимо скопировать в адресное пространство ядра. Можно выделить для этого память в куче, но доступ к куче относительно медленный. Чаще всего системному вызову poll передаётся не большое количество дескрипторов, поэтому часть массива можно разместить на стеке, а то, что не поместилось - в куче. Для этого используется связный список.
Элемент связного списка описывается структурой:

struct poll_list {
 struct poll_list *next;
 int len;
 struct pollfd entries[0];
};

struct poll_list *next - указатель на следующий элемент списка;
int len - длина массива дескрипторов;
struct pollfd entries[0] - указатель на массив;

Структура pollfd:

struct pollfd {
 int fd;
 short events;
 short revents;
};

Для указателя на массив дескрипторов используется массив нулевой длины. Почему не struct pollfd* entries будет понятно ниже.

Далее имеем вот такой код (несколько упрощённый вариант)

#define POLL_STACK_ALLOC 256

/*Количество элементов массива pollfd, которое поместится на стеке
  Также учитывается размер элемента списка, т.к. он тоже размещается на стеке*/

#define N_STACK_PPS ((sizeof(stack_pps) - sizeof(struct poll_list))  / \
   sizeof(struct pollfd))

#define POLLFD_PER_PAGE  ((PAGE_SIZE-sizeof(struct poll_list)) / sizeof(struct pollfd))


int do_sys_poll(struct pollfd *ufds, unsigned int nfds,
 struct timespec *end_time)
{
 struct poll_wqueues table;
 int err = -EFAULT, fdcount, len, size;

 /*Память под элемент списка и массив pollfd*/
 long stack_pps[POLL_STACK_ALLOC/sizeof(long)];
 /*Указатель на первый элемент списка*/
 struct poll_list *const head = (struct poll_list *)stack_pps;
 /*Указатель для прохода по списку*/
 struct poll_list *walk = head;
 /*Сколько ещё элементов необходимо скопировать*/
 unsigned long todo = nfds;

 /*При перво проходе цикла len содержит количество элементов,
    которые будут размещены на стеке*/
 len = min_t(unsigned int, nfds, N_STACK_PPS);
 for (;;) {
  walk->next = NULL;
  walk->len = len;
  if (!len)
   break;

  /*Копируем данные из пространства пользователя в пространство ядра*/
  if (copy_from_user(walk->entries, ufds + nfds-todo,
     sizeof(struct pollfd) * walk->len))
   goto out_fds;

  /*Сколько ещё осталось скопировать*/
  todo -= walk->len;
  if (!todo)
   break;

  /*Все элементы, которые не поместились на стеке, мы размещаем в куче
     постранично*/

  len = min(todo, POLLFD_PER_PAGE);
  size = sizeof(struct poll_list) + sizeof(struct pollfd) * len;
  walk = walk->next = kmalloc(size, GFP_KERNEL);
  if (!walk) {
   err = -ENOMEM;
   goto out_fds;
  }
 }

 ................................

 return err;
}

Макрос N_STACK_PPS вычисляет количество элементов + размер элемента списка, которые  могут поместиться на стеке. Получается, что в массиве stack_pps первые sizeof(poll_list) байт отводится под элемент списка, а sizeof(stack_pps) - sizeof(poll_list) байт под массив pollfd.
Надо как то ссылаться на массив pollfd из элемента списка. Известно, что массив располагается сразу после структуры poll_list в массиве stack_pps и в выделяемой в куче памяти. Для того, чтобы получить указатель на область сразу за poll_list используется массив нулевой длинны struct pollfd entries[0]. Это эффективней чем указатель struct pollfd* entries, т.к. он съел бы дополнительные байты памяти, и его необходимо было бы инициализировать адресом первого элемента массива.

Использование epoll

Мультиплексирование ввода-вывода

Существует множество способов написать сервер, обрабатывающий клиентский запросы. Вот некоторые из них:

• Последовательная обработка каждого запроса;
• Один процесс/поток на клиента;
• Мультиплексирование ввода-вывода.

Каждый подход обладает своими достоинствами и недостатками.

Я опишу мультиплексирование ввода вывода.
Принцип его работы в следующем: дескрипторы ввода/вывода помещается в список ожидания, затем процесс блокируется до наступления одного из событий:

Для читающего сокета:

• Можно выполнить чтение данных. Т.е. количество байт в буфере приёмнике больше или равно значению минимального количества данных (по умолчанию это значение равно 1, но его можно поменять с помощью параметра сокета SO_RCVLOWAT);
• На противоположном конце соединение закрывается;
• Сокет является прослушиваемым и число установленных соединений больше нуля.
• Ошибка сокета;

Для пишущего сокета:

• Можно выполнить запись данных. Т.е. количество байт в буфере передатчике больше или равно значению минимального количества данных (по умолчанию это значение равно 2048, но его можно поменять с помощью параметра сокета SO_SNDLOWAT);
• На противоположном конце соединение закрывается. При этом если записать данные в сокет, сгенерируется сигнал SIGPIPE;
• Ошибка сокета;

Также процесс выходит из состояния блокировки при получении внеполосных данных.

epoll
Системные вызовы группы epoll присущи только Linux. Они появились в ядре версии 2.6 и призваны исправить проблемы системных вызовов poll и select.
Работа с epoll осуществляется в три этапа:
1. Создаём дескриптор epoll, вызвав функцию

int epoll_create(int size);

2. Добавляем дескриптор ввода/вывода в массив ожидания, вызвав функцию

int epoll_ctl(int efd, int op, int fd, struct epool_event* event);

3. Запускаем ожидание готовности, вызвав функцию

int epoll_wait(int efd, struct epoll_event *events, int maxevents, int timeout);

За более подробной информацией по параметрам и возвращающим значениям функций, необходимо обратиться к документации.

Вот небольшой пример кода, использующий epoll:

#include <stdio.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>
#include <fcntl.h>
#include <strings.h>
#include <string.h>
#include <arpa/inet.h>
#include <errno.h>
#include <stdlib.h>
#include <sys/epoll.h>
#include <signal.h>
#include <iostream>

namespace
{

 int setnonblocking(int sock);
 void do_read(int fd);
 void do_write(int fd);
 void process_error(int fd);

}

int main(int argc, char* argv[])
{
 const int MAX_EPOLL_EVENTS = 100;
 const int BACK_LOG = 100;

 if (argc < 2)
 {
  std::cout <<  "Usage: server [port]" << std::endl;
  return 0;
 }

 char* p;
 int serv_port = strtol(argv[1], &p, 10);
 if (*p)
 {
  std::cout << "Invalid port number" << std::endl;
  return -1;
 }

 // Игнорируем сигнал ошибки работы с сокетом
 signal(SIGPIPE, SIG_IGN);

 // Создание дескриптора epoll
 int efd = epoll_create(MAX_EPOLL_EVENTS);

 int listenfd;
 listenfd = socket(AF_INET, SOCK_STREAM, 0);
 if (listenfd < 0)
 {
  perror("Socket creation");
  return -1;
 }

 // Неблокирующий режим
 setnonblocking(listenfd);

 struct sockaddr_in servaddr;
 bzero(&servaddr, sizeof(servaddr));
 servaddr.sin_family = AF_INET;
 servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
 servaddr.sin_port = htons(serv_port);

 if (bind(listenfd, (struct sockaddr*)&servaddr, sizeof(servaddr)) < 0)
 {
  perror("Socket bind");
  return -1;
 }

 if (listen(listenfd, BACK_LOG) < 0)
 {
  perror("Socket listen");
  return -1;
 }

 // Добавляем дескриптор в массив ожидания
 struct epoll_event listenev;
 listenev.events = EPOLLIN | EPOLLPRI | EPOLLET;
 listenev.data.fd = listenfd;
 if (epoll_ctl(efd, EPOLL_CTL_ADD, listenfd, &listenev) < 0)
 {
  perror("Epoll fd add");
  return -1;
 }

 socklen_t client;
 // Массив готовых дескрипторов
 struct epoll_event events[MAX_EPOLL_EVENTS];
 struct epoll_event connev;
 struct sockaddr_in cliaddr;

 int events_cout = 1;

 for (;;)
 {
  // Блокирование до готовности одно или нескольких дескрипторов
  int nfds = epoll_wait(efd, events, MAX_EPOLL_EVENTS, -1);

  for (int n = 0; n < nfds; ++n)
  {
   // Готов слушающий дескриптор
   if (events[n].data.fd == listenfd)
   {
    client = sizeof(cliaddr);
    int connfd = accept(listenfd, (struct sockaddr*) &cliaddr, &client);
    if (connfd < 0)
    {
     perror("accept");
     continue;
    }

    // Недостаточно места в массиве ожидания
    if (events_cout == MAX_EPOLL_EVENTS-1)
    {
     std::cout << "Event array is full" << std::endl;
     close(connfd);
     continue;
    }

    // Добавление клиентского дескриптора в массив ожидания
    setnonblocking(connfd);
    connev.data.fd = connfd;
    connev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLRDHUP;
    if (!epoll_ctl(efd, EPOLL_CTL_ADD, connfd, &connev) < 0)
    {
     perror("Epoll fd add");
     close(connfd);
     continue;
    }

    ++events_cout;
   }
   // Готов клиентский дескриптор
   else
   {
    // Выполням работу с дескриптором
    int fd = events[n].data.fd;

    if (events[n].events & EPOLLIN)
     do_read(fd);

    if (events[n].events & EPOLLOUT)
     do_write(fd);

    if (events[n].events & EPOLLRDHUP)
     process_error(fd);

    // В даннoм примере дескриптор просто закрывается и удаляется из массива ожидания.
    // В зависимости от логики работы можно не удалять дескриптор и подождать следующую порцию данных
    epoll_ctl(efd, EPOLL_CTL_DEL, fd, &connev);
    --events_cout;
    close(fd);
   }
  }
 }

 return 0;
}

namespace
{

 int setnonblocking(int sock)
 {
  int opts;

  opts = fcntl(sock,F_GETFL);
  if (opts < 0)
  {
   perror("fcntl(F_GETFL)");
   return -1;
  }
  opts = (opts | O_NONBLOCK);
  if (fcntl(sock,F_SETFL,opts) < 0)
  {
   perror("fcntl(F_SETFL)");
   return -1;
  }

  return 0;
 }

 void do_read(int fd)
 {
  std::cout << "do_read" << std::endl;
 }

 void do_write(int fd)
 {
  std::cout << "do_write" << std::endl;
 }

 void process_error(int fd)
 {
  std::cout << "process_error" << std::endl;
 }
}

USB драйвер в Linux

Введение

Драйвер устройства - это некий код, который позволяет операционной системе взаимодействовать с устройством. Ядро Linux предоставляет API для работы с физическими интерфейсами (для USB это подсистема ядра USB Core). Этот слой можно назвать транспортным уровнем. Прикладной уровень протоколов (например HID для работы с USB мышью) реализуется драйверами устройств.
В ОС Linux драйвера устройств выполняются в виде модулей ядра. Модулем ядра называют часть кода ядра, который может динамически загружаться и выгружаться во время работы ядра.

Пишем модуль

Примерный код драйвера для usb находится в файле drivers/usb/usb-skeleton.c дерева исходного кода ядра.

У каждого модуля ядра есть точка входа и точка выхода. Это функции, которые вызываются при загрузки и выгрузки модуля соответственно.

Для указания точки входа модуля ядра, используется макрос

module_init(my_driver_init);

Прототип функции инициализации модуля -

int __init my_driver_init(void);

В функции выполняется инициализация структур, регистрация устройств и т.д.
Для регистрации драйвера в системе имеется структура usb_driver, в которой указывается имя модуля, параметры работы и указатели на функции обратного вызова драйвера.
Здесь как и во многих подсистемах ядра (например в виртуальной файловой системе), используется объектно ориентированный подход.
Для нашего модуля структура будет выглядеть так:

static struct usb_driver my_driver = {
    .name  = "my_driver",
    .probe  = my_driver_probe,
    .disconnect = my_driver_disconnect,
    .id_table = my_driver_id_table
};

Поле name - имя драйвера. Оно должно быть уникальным и обычно совпадает с именем модуля.
Поле probe - указатель на функцию, которую подсистема USB ядра вызывает при подключении устройства.
Поле disconnect - указатель на функцию, которую подсистема USB ядра вызывает при отключении устройства.
Поле id_table - указатель на структуру usb_device_id (см. ниже).

Функция

my_driver_init()

в нашем случае выглядит вот так:

static int __init my_driver_init(void)
{
    int result;

    printk("My driver init");

    result = usb_register(&my_driver);
    if (result)
        err("My driver register failed with error number %d", result);

    return result;
}

В структуре usb_device_id указываются идентификаторы производителя и устройства, для которых предназначен драйвер. При подключении устройства подсистема USB ядра считывает эти идентификатор с устройства и вызывает соответствующий драйвер.
В нашем случае определение структуры будет следующим:

#define VENDOR_ID  0xfff0
#define PRODUCT_ID  0xfff0

static const struct usb_device_id my_driver_id_table[] = {
    { USB_DEVICE(VENDOR_ID, PRODUCT_ID) },
    { }
};

В реальном драйвере вместо кода 0xfff0 должны стоять идентификаторы реального устройства, которые можно получить у производителя.

Функция probe и disconnect в нашем модуле будут выглядеть следующим образом:

static int my_driver_probe(struct usb_interface* interface, const struct usb_device_id* id)
{
    printk("My moudle probe\n");
}

static void my_driver_disconnect(struct usb_interface* interface)
{
    printk("My driver disconnect\n");
}

Для указания точки выхода используется макрос

module_exit(my_driver_exit);

Прототип функции выхода -

void __exit my_driver_exit(void)

В функции выполняется очистка ресурсов.
В нашем случае она выглядит так:

static void __exit my_driver_exit(void)
{
    printk("My driver exit");
    usb_deregister(&my_driver);
}

Этот минимальный код ничего кроме реакции на подключение устройства пока не делает. В дальнейшем я постараюсь описать процесс обмена данными с устройством.

Литература
1. Р. Лав. Разработка ядра Linux.
2. Linux device drivers 3rd edition.

IP адрес интерфейса

Некоторое время назад передо мной встала задача инициировать процедуру получения IP адрес для интерфейса по DHCP. Для это существует команда dhclient <интерфейс>. Нужно как то понимать, что dhclient отработал корректно и IP адрес получен. Можно парсить вывод программы на sdtout, а можно проверить, установлен ли IP адрес у интерфейса. С парсингом вывода всё понятно, поэтому я расскажу как получить IP адрес интерфейса.

Приведённый ниже код получает список сетевых интерфейсов, и выводит их IP адреса.

#include <sys/ioctl.h>
#include <net/if.h>
#include <arpa/inet.h>

#include <iostream>

int main (int argc, char* argv[])
{
    struct ifconf ifc;
    int numreqs = 30;

    int skfd = socket(AF_INET, SOCK_DGRAM, 0);

    //Получаем список интерфейсов
    ifc.ifc_buf = 0;
    for (;;)
    {
        ifc.ifc_len = sizeof(struct ifreq) * numreqs;
        ifc.ifc_buf = new char[ifc.ifc_len];

        if (ioctl(skfd, SIOCGIFCONF, &ifc) < 0)
        {
            return false;
        }
        if (ifc.ifc_len == sizeof(struct ifreq) * numreqs)
        {
            delete[] ifc.ifc_buf;
            numreqs += 10;
            continue;
        }
        break;
    }

    //Проходимся по списку интерфейсов
    bool res = false;
    struct ifreq *ifr = ifc.ifc_req;
    for (int n = 0; n < ifc.ifc_len; n += sizeof(struct ifreq))
    {
        char addr[16];
        sockaddr_in* sock = reinterpret_cast<sockaddr_in*>(&ifr->ifr_ifru.ifru_addr);
        inet_ntop(AF_INET, &sock->sin_addr, addr, 16);

        std::cout << ifr->ifr_name << " - " << addr << std::endl;

        ifr++;
    }

    return 0;
}