Огромное падение производительности в кассандра-орме после миллиона записей

Question

Я использую плагин cassandra-orm (cassandra-orm: 0.4.5) для переноса кликов из Postgres DB в Cassandra. (Я знаю, что мог бы использовать импорт необработанных данных, но я хочу использовать groupBy и явные индексы, поддерживаемые плагином).

Процедура миграции проста: я выбираю несколько кликов из Postgres (через GORM) и затем сбрасываю их в Cassandra. Каждый щелчок - это новая запись, и новый объект создается в Grails и сохраняется в Cassandra. С 20 потоками я смог достичь пропускной способности 2000 кликов в секунду. После импорта 5 миллионов кликов производительность начала резко снижаться до 50 кликов в секунду.

Я сделал некоторое профилирование и обнаружил, что 19 потоков ожидают (припаркованы), и один поток выполняет перефразировку в Groovy AbstractConcurrentMapBase.

трассировка стека для ожидающих потоков:

Name: pool-4-thread-2
State: WAITING on org.codehaus.groovy.util.ManagedConcurrentMap$Segment@5387f7af
Total blocked: 45,027  Total waited: 55,891

Stack trace: 
 sun.misc.Unsafe.park(Native Method)
java.util.concurrent.locks.LockSupport.park(LockSupport.java:156)
java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:811)
java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireQueued(AbstractQueuedSynchronizer.java:842)
java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:1178)
org.codehaus.groovy.util.LockableObject.lock(LockableObject.java:34)
org.codehaus.groovy.util.AbstractConcurrentMap$Segment.put(AbstractConcurrentMap.java:101)
org.codehaus.groovy.util.AbstractConcurrentMap$Segment.getOrPut(AbstractConcurrentMap.java:97)
org.codehaus.groovy.util.AbstractConcurrentMap.getOrPut(AbstractConcurrentMap.java:35)
org.codehaus.groovy.runtime.metaclass.ThreadManagedMetaBeanProperty$ThreadBoundGetter.invoke(ThreadManagedMetaBeanProperty.java:180)
groovy.lang.MetaMethod.doMethodInvoke(MetaMethod.java:233)
groovy.lang.MetaClassImpl.getProperty(MetaClassImpl.java:1604)
groovy.lang.ExpandoMetaClass.getProperty(ExpandoMetaClass.java:1140)
groovy.lang.MetaClassImpl.getProperty(MetaClassImpl.java:3332)
groovy.lang.ExpandoMetaClass.getProperty(ExpandoMetaClass.java:1152)
com.nosql.Click.getProperty(Click.groovy)

трассировка стека для перефразирующего потока:

Name: pool-4-thread-11
State: RUNNABLE
Total blocked: 46,544  Total waited: 57,433

Stack trace: 
 org.codehaus.groovy.util.AbstractConcurrentMapBase$Segment.rehash(AbstractConcurrentMapBase.java:217)
org.codehaus.groovy.util.AbstractConcurrentMap$Segment.put(AbstractConcurrentMap.java:105)
org.codehaus.groovy.util.AbstractConcurrentMap$Segment.getOrPut(AbstractConcurrentMap.java:97)
org.codehaus.groovy.util.AbstractConcurrentMap.getOrPut(AbstractConcurrentMap.java:35)
org.codehaus.groovy.runtime.metaclass.ThreadManagedMetaBeanProperty$ThreadBoundGetter.invoke(ThreadManagedMetaBeanProperty.java:180)
groovy.lang.MetaMethod.doMethodInvoke(MetaMethod.java:233)
groovy.lang.MetaClassImpl.getProperty(MetaClassImpl.java:1604)
groovy.lang.ExpandoMetaClass.getProperty(ExpandoMetaClass.java:1140)
groovy.lang.MetaClassImpl.getProperty(MetaClassImpl.java:3332)
groovy.lang.ExpandoMetaClass.getProperty(ExpandoMetaClass.java:1152)
com.fma.nosql.Click.getProperty(Click.groovy)

После нескольких часов отладки я обнаружил, что проблема в динамическом свойстве "_cassandra_cluster_", которое добавляется ко всем управляемым объектам плагина:

// cluster property (_cassandra_cluster_)
clazz.metaClass."${CLUSTER_PROP}" = null

Затем это свойство внутренне сохраняется в карте ThreadManagedMetaBeanProperty instance2Prop. Когда к динамическому свойству обращаются def cluster = click._cassandra_cluster_ затем экземпляр клика сохраняется на карту instance2Prop с мягкой ссылкой. Пока все хорошо, мягкие ссылки можно собирать, верно. Однако в реализации ManagedConcurrentMap, похоже, есть ошибка, которая игнорирует элементы, собираемые мусором, и продолжает перефразировать и расширять карту (описано здесь и здесь).

Временное решение

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

public void rehashClickSegmentsIfNecessary() {
    ManagedConcurrentMap instanceMap = lookupInstanceMap(Click.class, "_cassandra_cluster_")
    if(instanceMap.fullSize() - instanceMap.size() > 50000) {
        //we have more than 50 000 zombie references in map
        rehashSegments(instanceMap)
    }
}

private void rehashSegments(ManagedConcurrentMap instanceMap) {
    org.codehaus.groovy.util.ManagedConcurrentMap.Segment[] segments = instanceMap.segments
    for(int i=0;i<segments.length;i++) {
        segments[i].lock()
        try {
            segments[i].rehash()
        } finally {
            segments[i].unlock()
        }
    }
}

private ManagedConcurrentMap lookupInstanceMap(Class clazz, String prop) {
    MetaClassRegistry registry = GroovySystem.metaClassRegistry
    MetaClassImpl metaClass = registry.getMetaClass(clazz)
    return metaClass.getMetaProperty(prop, false).instance2Prop
}

Есть ли у вас опыт работы с cassandra-orm или любым другим плагином Grails, подключающимся к cassandra?