Вступление

Привет, меня зовут Денис Агапитов, я руководитель группы Platform Core компании Bercut. Работаю в компании без малого 20 лет, из них 18 пишу на Java.

Сегодня я расскажу об опыте увеличения производительности сетевого стэка и проблемах, с которыми можно столкнуться при использовании NIO в Java.

Эта статья основана на реальной практике борьбы с "OutOfMemory: direct memory" в шине данных гибридной интеграционной платформы.

Группа Platform Core, которой я руковожу, занимается разработкой и развитием гибридной интеграционной платформы, поддержкой систем и сервисов, написанных на платформе.

Платформа включает в себя:

• Шину данных ESB.

• Приложения API Gateway, SLES (сервер исполнения бизнес-процессов), SA Container (сервер с сервисами на Java), Notification Broker.

• Платформенные сервисы: Scheduler, Service Profile Management и прочие.

• Поддержку интеграции со Spring.

Итак, начнём с предпосылок, которые подвигли заняться анализом данной проблемы.

Особенности работы шины данных в Bercut

Наша гибридная интеграционная платформа имеет свою транспортную шину (RTSIB). Это ESB (enterprise service bus) в рамках архитектуры SOA (service-oriented architecture) со своими стеками HTTP и проприетарного асинхронного протокола RTSIB.

По своей сути это mesh-сеть между разными узлами и приложениями платформы.

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

HTTP соединения (на текущий момент мы поддерживаем версию 1.x) в виду синхронной архитектуры не требуют большого количества потоков обслуживания, потоки для них достаются из пула - примерно один поток на 25 соединений.

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

Особенности работы с сетью TCP/IP в неблокирующем режиме в Java

Кто уже интересовался тем, как работает запись в сокет на Java или просто любит смотреть исходные коды JDK, вероятно знает основные особенности работы с сокетом, но для понимания проблемы предлагаю ещё раз их проговорить.

Из исходных кодов Open JDK 13 (основная версия, используемой у нас Java) видно, что если записываемый ByteBuffer является DirectByteBuffer, то запись происходит сразу (writeFromNativeBuffer), а если он расположен в Heap, то сначала достаётся временный DirectByteBuffer, производится копирование и запись из временного DirectByteBuffer.

Код записи в сокет из Open JDK 13 ( IOUtil.java ):

static int write(FileDescriptor fd, ByteBuffer src, long position,
                 boolean directIO, int alignment, NativeDispatcher nd)
    throws IOException
{
    if (src instanceof DirectBuffer) {
        return writeFromNativeBuffer(fd, src, position, directIO, alignment, nd);
    }
 
    // Substitute a native buffer
    int pos = src.position();
    int lim = src.limit();
    assert (pos <= lim);
    int rem = (pos <= lim ? lim - pos : 0);
    ByteBuffer bb;
    if (directIO) {
        Util.checkRemainingBufferSizeAligned(rem, alignment);
        bb = Util.getTemporaryAlignedDirectBuffer(rem, alignment);
    } else {
        bb = Util.getTemporaryDirectBuffer(rem);
    }
    try {
        bb.put(src);
        bb.flip();
        // Do not update src until we see how many bytes were written
        src.position(pos);
 
        int n = writeFromNativeBuffer(fd, bb, position, directIO, alignment, nd);
        if (n > 0) {
            // now update src
            src.position(pos + n);
        }
        return n;
    } finally {
        Util.offerFirstTemporaryDirectBuffer(bb);
    }
}

Дополнительно осложняет ситуацию то, что внутри реализации JDK имеется КЭШ DirectByteBuffer с привязкой к потоку (ThreadLocal):

private static ThreadLocal<BufferCache> bufferCache = new TerminatingThreadLocal<>() {
    @Override
    protected BufferCache initialValue() {
        return new BufferCache();
    }
    @Override
    protected void threadTerminated(BufferCache cache) { // will never be null
        while (!cache.isEmpty()) {
            ByteBuffer bb = cache.removeFirst();
            free(bb);
        }
    }
};
 
public static ByteBuffer getTemporaryDirectBuffer(int size) {
    // If a buffer of this size is too large for the cache, there
    // should not be a buffer in the cache that is at least as
    // large. So we'll just create a new one. Also, we don't have
    // to remove the buffer from the cache (as this method does
    // below) given that we won't put the new buffer in the cache.
    if (isBufferTooLarge(size)) {
        return ByteBuffer.allocateDirect(size);
    }
 
    BufferCache cache = bufferCache.get();
    ByteBuffer buf = cache.get(size);
    if (buf != null) {
        return buf;
    } else {
        // No suitable buffer in the cache so we need to allocate a new
        // one. To avoid the cache growing then we remove the first
        // buffer from the cache and free it.
        if (!cache.isEmpty()) {
            buf = cache.removeFirst();
            free(buf);
        }
        return ByteBuffer.allocateDirect(size);
    }
}

И после каждого использования временного DirectByteBuffer, он помещается в КЭШ. При этом, если в КЭШе нет DirectByteBuffer необходимого размера, он аллоцируется и после использования также помещается в КЭШ ( Util.offerFirstTemporaryDirectBuffer(bb) ).

Суть проблемы

В первых версиях платформы мы просто использовали HeapByteBuffer через простую и понятную static-функцию ByteBuffer.wrap(byte[] data) и бед вроде как не знали.

Всё работало, скорость была достаточная для текущих telecom-сервисов, работающих на платформе, но в один прекрасный день размер данных DWH (Data Warehouse), проходящих через нашу шину достиг критического объёма в мегабайтах и мы получили OOM Direct Memory.

Почему же так произошло? А вот почему: как обозначил выше, мы имеем mesh-сеть с множеством обслуживающих потоков и имеем данные большого размера, проходящие через эти потоки, которые складывают off-heap память в ThreadLocal КЭШи этих потоков. Достигнув предела насыщения использования off-heap памяти  мы получаем OOM. Конечно, первым действием было увеличение параметра запуска JVM "-XX:MaxDirectMemorySize". Размер используемой direct памяти пришлось увеличить, потом увеличить ещё. Это стало тем самым звоночком, что с проблемой надо разбираться и как можно скорее.

Анализ возможных путей решения

После осознания проблемы, мы провели анализ возможных путей её решения и нашли следующие варианты:

• Писать в цикле блоками, сдвигая вручную position и limit в записываемом ByteBuffer. Это должно помочь, так как в IOUtils временный DirectByteBuffer выделяется размером size = limit - position.

• Перейти на использование ByteBuffer.allocateDirect().

• Написать промежуточную абстракцию, содержащую нарезку из ByteBuffer одного размера, где ByteBuffer одного размера берутся из общего пула и после использования возвращаются обратно.

Погружаясь всё глубже в исследование проблемы стало понятно, что необходимо провести сравнительное тестирование разных размеров ByteBuffer и разных вариантов их использования для чтения и записи из сокета.

За несколько часов я написал тестовое приложение, которое эмулирует 4 вида работы с сокетом:

• Аллоцирование HeapByteBuffer при каждой записи/чтении.

• Аллоцирование DirectByteBuffer при каждой записи/чтении.

• Переиспользование HeapByteBuffer при каждой записи/чтении.

• Переиспользование DirectByteBuffer при каждой записи/чтении.

В тестовом приложении отсутствует маршаллинг (заполнение реальными данными), а присутствует только работа по записи и чтению из сокета с разными вариантами использования ByteBuffer.

Здесь приводить исходные коды не буду, но кто желает может ознакомится с ними на Github.

На выходе мы получили такую картину:

Из результатов видно, что до 1Mb самым медленным вариантом является аллокация DirectByteBuffer. Аллокация HeapByteBuffer через wrap и кэшированный HeapByteBuffer примерно равны с небольшим лидерством кэшированного. Из общей картины выбивается кэшированный DirectByteBuffer, что логично, так как пишется он напрямую, а время на аллокацию отсутствует.

Выбор и реализация

Для реализации выбрали 3 вариант решения проблемы: написать промежуточную абстракцию, содержащую нарезку из DirectByteBuffer одного размера, которые берутся и возвращаются в общий пул. За основной размер части пакета (размер DirectByteBuffer) было выбрано значение в 32Kb как минимальный по размеру пик при тестировании пропускной способности. Безусловно, так как у нас реализован и стэк HTTPs, фабрика может отдавать и пул с отличными от 32Kb размерами DirectByteBuffer, опираясь на PacketBufferSize и ApplicationBufferSize из настроек текущей сессии SSLEngine.

При написании слоя абстракции, названной CompositeBuffer, конечно же я реализовал и Input/Output streams, работающие напрямую с CompositeBuffer. Это было необходимо для нормальной работы слоя marshalling/unmarshalling.

В качестве хранилища уже аллоцированных DirectByteBuffer сделал простой стэк на CAS механизме:

public class CasStack<L extends LinkedObject<L>> {
 
    public interface LinkedObject<L extends LinkedObject> {
     
        public L getNext();
         
        public void setNext(L next);
         
    }
 
    private final AtomicReference<L> head = new AtomicReference<>();
 
    public void add(L lo) {
        for (;;) {
            lo.setNext(head.get());
            if (head.compareAndSet(lo.getNext(), lo)) {
                return;
            }
        }
    }
 
    public L poll() {
        L lo;
        for (;;) {
            lo = head.get();
            if (lo == null) {
                return null;
            }
            if (head.compareAndSet(lo, lo.getNext())) {
                lo.setNext(null);
                return lo;
            }
        }
 
    }
 
}

А примерно вот так выглядит часть основного класса CompositeBuffer в разрезе работы с чтением из сокета и записью в сокет (код был адаптирован для статьи):

DirectByteBuffer[] buffers;
int pos;
 
@Override
public int getBufIndex(int position) {
    return position / pool.getPartCapacity();
}
 
@Override
public int read(ReadableByteChannel channel) throws IOException {
    ensureCapacity();
    int cur = getBufIndex(pos), readed = 0, read;
    for (;;) {
        try {
            if (hasRemaining()) {
                if (buffers[cur].hasRemaining()) {
                    read = channel.read(containers[cur]);
                    if (read < 0) {
                        return readed > 0 ? readed : read;
                    }
                    pos += read;
                    readed += read;
                    if (containers[cur].hasRemaining()) {
                        return readed;
                    }
                }
                cur++;
                if (cur == containers.length) {
                    expandCapacity();
                }
            } else {
                return readed;
            }
        } catch (IOException) {
            if (readed > 0) {
                return readed;
            }
            throw ex;
        }
    }
}
 
@Override
public int write(WritableByteChannel channel) throws IOException {
    int cur = getBufIndex(pos), writed = 0, write;
    for (;;) {
        try {
            if (hasRemaining()) {
                if (buffers[cur].hasRemaining()) {
                    write = channel.write(buffers[cur]);
                    if (write < 0) {
                        return writed > 0 ? writed : write;
                    }
                    pos += write;
                    writed += write;
                    if (buffers[cur].hasRemaining()) {
                        return writed;
                    }
                }
                cur++;
                if (cur == containers.length) {
                    if (writed == 0 && hasRemaining()) {
                        release();
                        throw new CompositeBufferLifecycleError();
                    }
                    return writed;
                }
            } else {
                return writed;
            }
        } catch (IOException ex) {
            if (writed > 0) {
                return writed;
            }
            throw ex;
        }
    }
}

Конечно же пришлось написать далеко не один класс, а ещё несколько уровней абстракции, таких, как DirectContainer и механизмы addRef/releaseRef, проверку на ошибки жизненного цикла всей библиотеки и многое другое.

Заключение

По завершению оптимизации и переходу на переиспользуемый DirectByteBuffer пропускная способность шины увеличилась примерно вдвое.

До данной доработки размер off-heap памяти мог достигать 1-3Gb и складывался из максимальных размеров сообщений, прошедших через каждое соединение.

Сейчас же потребление off-heap памяти пулами довольно скромное - на среднем сервисе оно составляет всего 10-20 Mb.

На более сложном компоненте с парой сотен входящих вызовов в секунду, которые порождают до 15 тысяч внутренних вызовов на каждый входящий вызов - размер off-heap пула занимает менее 100Mb.

Важно ещё и то, что теперь off-heap память, используемая приложениями для работы с сетью более контролируема и растёт не от корреляции с количеством соединений (потоков), а только от корреляции с количеством сообщений проходящих через шину в единицу времени и средним размером сообщения.

При этом надо понимать, что размер off-heap памяти не может быть меньше максимального размера сообщения, когда-либо проходившего через узел.

Конечно, полностью исключить OOM direct memory таким решением всё равно не получится, но теперь off-heap память можно прогнозировать.