Use separate flush targets in attribute mangager to shrink attribute vectors.

6 files changed, 69 insertions, 43 deletions

diff --git a/searchcore/src/tests/proton/attribute/attributeflush_test.cpp b/searchcore/src/tests/proton/attribute/attributeflush_test.cpp
index 5375f7817ec..714e9076d16 100644
--- a/searchcore/src/tests/proton/attribute/attributeflush_test.cpp
+++ b/searchcore/src/tests/proton/attribute/attributeflush_test.cpp
@@ -6,6 +6,7 @@
 #include <vespa/searchcore/proton/attribute/attributedisklayout.h>
 #include <vespa/searchcore/proton/attribute/attribute_writer.h>
 #include <vespa/searchcore/proton/attribute/flushableattribute.h>
+#include <vespa/searchcore/proton/flushengine/shrink_lid_space_flush_target.h>
 #include <vespa/searchlib/attribute/attributefactory.h>
 #include <vespa/searchlib/attribute/integerbase.h>
 #include <vespa/searchlib/common/indexmetainfo.h>
@@ -389,9 +390,11 @@ Test::requireThatFlushTargetsCanBeRetrieved(void)
     f.addAttribute("a4");
     f.addAttribute("a5");
     std::vector<IFlushTarget::SP> ftl = am.getFlushTargets();
-    EXPECT_EQUAL(2u, ftl.size());
+    EXPECT_EQUAL(4u, ftl.size());
     EXPECT_EQUAL(am.getFlushable("a4").get(), ftl[0].get());
-    EXPECT_EQUAL(am.getFlushable("a5").get(), ftl[1].get());
+    EXPECT_EQUAL(am.getShrinker("a4").get(),  ftl[1].get());
+    EXPECT_EQUAL(am.getFlushable("a5").get(), ftl[2].get());
+    EXPECT_EQUAL(am.getShrinker("a5").get(),  ftl[3].get());
 }
 
 
@@ -528,7 +531,7 @@ Test::requireThatShrinkWorks()
     
     av->addDocs(1000 - av->getNumDocs());
     av->commit(10, 10);
-    IFlushTarget::SP ft = am.getFlushable("a10");
+    IFlushTarget::SP ft = am.getShrinker("a10");
     EXPECT_EQUAL(ft->getApproxMemoryGain().getBefore(),
                  ft->getApproxMemoryGain().getAfter());
     AttributeGuard::UP g = am.getAttribute("a10");
diff --git a/searchcore/src/tests/proton/documentdb/document_subdbs/document_subdbs_test.cpp b/searchcore/src/tests/proton/documentdb/document_subdbs/document_subdbs_test.cpp
index abfd80c9582..0b4b9ba992f 100644
--- a/searchcore/src/tests/proton/documentdb/document_subdbs/document_subdbs_test.cpp
+++ b/searchcore/src/tests/proton/documentdb/document_subdbs/document_subdbs_test.cpp
@@ -685,23 +685,25 @@ assertTarget(const vespalib::string &name,
 TEST_F("require that flush targets can be retrieved", FastAccessFixture)
 {
 	IFlushTarget::List targets = getFlushTargets(f);
-	EXPECT_EQUAL(4u, targets.size());
+	EXPECT_EQUAL(5u, targets.size());
 	EXPECT_EQUAL("subdb.attribute.attr1", targets[0]->getName());
-	EXPECT_EQUAL("subdb.documentmetastore", targets[1]->getName());
-	EXPECT_EQUAL("subdb.summary.compact", targets[2]->getName());
-	EXPECT_EQUAL("subdb.summary.flush", targets[3]->getName());
+	EXPECT_EQUAL("subdb.attributeshrink.attr1", targets[1]->getName());
+	EXPECT_EQUAL("subdb.documentmetastore", targets[2]->getName());
+	EXPECT_EQUAL("subdb.summary.compact", targets[3]->getName());
+	EXPECT_EQUAL("subdb.summary.flush", targets[4]->getName());
 }
 
 TEST_F("require that flush targets can be retrieved", SearchableFixture)
 {
 	IFlushTarget::List targets = getFlushTargets(f);
-	EXPECT_EQUAL(6u, targets.size());
+	EXPECT_EQUAL(7u, targets.size());
 	EXPECT_TRUE(assertTarget("subdb.attribute.attr1", FType::SYNC, FComponent::ATTRIBUTE, *targets[0]));
-	EXPECT_TRUE(assertTarget("subdb.documentmetastore", FType::SYNC, FComponent::ATTRIBUTE, *targets[1]));
-	EXPECT_TRUE(assertTarget("subdb.memoryindex.flush", FType::FLUSH, FComponent::INDEX, *targets[2]));
-	EXPECT_TRUE(assertTarget("subdb.memoryindex.fusion", FType::GC, FComponent::INDEX, *targets[3]));
-	EXPECT_TRUE(assertTarget("subdb.summary.compact", FType::GC, FComponent::DOCUMENT_STORE, *targets[4]));
-	EXPECT_TRUE(assertTarget("subdb.summary.flush", FType::SYNC, FComponent::DOCUMENT_STORE, *targets[5]));
+	EXPECT_TRUE(assertTarget("subdb.attributeshrink.attr1", FType::GC, FComponent::ATTRIBUTE, *targets[1]));
+	EXPECT_TRUE(assertTarget("subdb.documentmetastore", FType::SYNC, FComponent::ATTRIBUTE, *targets[2]));
+	EXPECT_TRUE(assertTarget("subdb.memoryindex.flush", FType::FLUSH, FComponent::INDEX, *targets[3]));
+	EXPECT_TRUE(assertTarget("subdb.memoryindex.fusion", FType::GC, FComponent::INDEX, *targets[4]));
+	EXPECT_TRUE(assertTarget("subdb.summary.compact", FType::GC, FComponent::DOCUMENT_STORE, *targets[5]));
+	EXPECT_TRUE(assertTarget("subdb.summary.flush", FType::SYNC, FComponent::DOCUMENT_STORE, *targets[6]));
 }
 
 TEST_F("require that only fast-access attributes are instantiated", FastAccessOnlyFixture)
diff --git a/searchcore/src/vespa/searchcore/proton/attribute/CMakeLists.txt b/searchcore/src/vespa/searchcore/proton/attribute/CMakeLists.txt
index cdb7b01254e..f0e6520d0b4 100644
--- a/searchcore/src/vespa/searchcore/proton/attribute/CMakeLists.txt
+++ b/searchcore/src/vespa/searchcore/proton/attribute/CMakeLists.txt
@@ -33,4 +33,5 @@ vespa_add_library(searchcore_attribute STATIC
     initialized_attributes_result.cpp
     sequential_attributes_initializer.cpp
     DEPENDS
+    searchcore_flushengine
 )
diff --git a/searchcore/src/vespa/searchcore/proton/attribute/attributemanager.cpp b/searchcore/src/vespa/searchcore/proton/attribute/attributemanager.cpp
index 03ec1836655..c0ac043ebb8 100644
--- a/searchcore/src/vespa/searchcore/proton/attribute/attributemanager.cpp
+++ b/searchcore/src/vespa/searchcore/proton/attribute/attributemanager.cpp
@@ -8,9 +8,11 @@
 #include "imported_attributes_repo.h"
 #include "sequential_attributes_initializer.h"
 #include "flushableattribute.h"
+#include <vespa/searchcore/proton/flushengine/shrink_lid_space_flush_target.h>
 #include <vespa/searchlib/attribute/attributecontext.h>
 #include <vespa/searchlib/attribute/interlock.h>
 #include <vespa/searchlib/common/isequencedtaskexecutor.h>
+#include <vespa/searchlib/common/threaded_compactable_lid_space.h>
 #include <vespa/searchlib/attribute/attributevector.h>
 #include <vespa/vespalib/io/fileutil.h>
 #include <vespa/vespalib/stllike/hash_map.hpp>
@@ -23,6 +25,7 @@ using search::AttributeContext;
 using search::AttributeEnumGuard;
 using search::AttributeGuard;
 using search::AttributeVector;
+using search::common::ThreadedCompactableLidSpace;
 using search::TuneFileAttributes;
 using search::attribute::IAttributeContext;
 using search::attribute::IAttributeVector;
@@ -60,6 +63,15 @@ bool matchingTypes(const AttributeVector::SP &av, const search::attribute::Confi
     }
 }
 
+search::SerialNum estimateShrinkSerialNum(const AttributeVector &attr)
+{
+    search::SerialNum serialNum = attr.getCreateSerialNum();
+    if (serialNum > 0) {
+        --serialNum;
+    }
+    return std::max(attr.getStatus().getLastSyncToken(), serialNum);
+}
+
 }
 
 AttributeManager::AttributeWrap::AttributeWrap(const AttributeVectorSP & a,
@@ -116,27 +128,27 @@ AttributeManager::internalAddAttribute(const AttributeSpec &spec,
     AttributeInitializerResult result = initializer.init();
     if (result) {
         result.getAttribute()->setInterlock(_interlock);
-        addAttribute(AttributeWrap::normalAttribute(result.getAttribute()));
+        SerialNum shrinkSerialNum = estimateShrinkSerialNum(*result.getAttribute());
+        addAttribute(AttributeWrap::normalAttribute(result.getAttribute()), shrinkSerialNum);
     }
     return result.getAttribute();
 }
 
 void
-AttributeManager::addAttribute(const AttributeWrap &attribute)
+AttributeManager::addAttribute(const AttributeWrap &attribute, SerialNum shrinkSerialNum)
 {
     LOG(debug, "Adding attribute vector '%s'", attribute.getAttribute()->getBaseFileName().c_str());
     _attributes[attribute.getAttribute()->getName()] = attribute;
     assert(attribute.getAttribute()->getInterlock() == _interlock);
     if ( ! attribute.isExtra() ) {
         // Flushing of extra attributes is handled elsewhere
-        auto flusher = std::make_shared<FlushableAttribute>
-                       (attribute.getAttribute(),
-                        _diskLayout->createAttributeDir(attribute.getAttribute()->getName()),
-                        _tuneFileAttributes,
-                        _fileHeaderContext,
-                        _attributeFieldWriter,
-                        _hwInfo);
-        auto shrinker = std::shared_ptr<ShrinkLidSpaceFlushTarget>();
+        auto attr = attribute.getAttribute();
+        const vespalib::string &name = attr->getName();
+        auto flusher = std::make_shared<FlushableAttribute>(attr, _diskLayout->createAttributeDir(name), _tuneFileAttributes, _fileHeaderContext, _attributeFieldWriter, _hwInfo);
+        auto shrinkwrap = std::make_shared<ThreadedCompactableLidSpace>(attr, _attributeFieldWriter, _attributeFieldWriter.getExecutorId(name));
+        using Type = IFlushTarget::Type;
+        using Component = IFlushTarget::Component;
+        auto shrinker = std::make_shared<ShrinkLidSpaceFlushTarget>("attributeshrink." + name, Type::GC, Component::ATTRIBUTE, shrinkSerialNum, attr);
         _flushables[attribute.getAttribute()->getName()] = FlushableWrap(flusher, shrinker);
         _writableAttributes.push_back(attribute.getAttribute().get());
     }
@@ -168,7 +180,12 @@ AttributeManager::transferExistingAttributes(const AttributeManager &currMgr,
         if (matchingTypes(av, aspec.getConfig())) { // transfer attribute
             LOG(debug, "Transferring attribute vector '%s' with %u docs and serial number %lu from current manager",
                        av->getName().c_str(), av->getNumDocs(), av->getStatus().getLastSyncToken());
-            addAttribute(AttributeWrap::normalAttribute(av));
+            auto wrap = currMgr.findFlushable(aspec.getName());
+            assert(wrap != nullptr);
+            auto shrinker = wrap->getShrinker();
+            assert(shrinker);
+            SerialNum shrinkSerialNum = shrinker->getFlushedSerialNum();
+            addAttribute(AttributeWrap::normalAttribute(av), shrinkSerialNum);
         } else {
             toBeAdded.push_back(aspec);
         }
@@ -208,7 +225,7 @@ AttributeManager::transferExtraAttributes(const AttributeManager &currMgr)
 {
     for (const auto &kv : currMgr._attributes) {
         if (kv.second.isExtra()) {
-            addAttribute(kv.second);
+            addAttribute(kv.second, 0);
         }
     }
 }
@@ -297,8 +314,10 @@ AttributeManager::addInitializedAttributes(const std::vector<AttributeInitialize
 {
     for (const auto &result : attributes) {
         assert(result);
-        result.getAttribute()->setInterlock(_interlock);
-        addAttribute(AttributeWrap::normalAttribute(result.getAttribute()));
+        auto attr = result.getAttribute();
+        attr->setInterlock(_interlock);
+        SerialNum shrinkSerialNum = estimateShrinkSerialNum(*attr);
+        addAttribute(AttributeWrap::normalAttribute(attr), shrinkSerialNum);
     }
 }
 
@@ -306,15 +325,16 @@ void
 AttributeManager::addExtraAttribute(const AttributeVector::SP &attribute)
 {
     attribute->setInterlock(_interlock);
-    addAttribute(AttributeWrap::extraAttribute(attribute));
+    addAttribute(AttributeWrap::extraAttribute(attribute), 0);
 }
 
 void
 AttributeManager::flushAll(SerialNum currentSerial)
 {
-    for (const auto &kv : _flushables) {
+    auto flushTargets = getFlushTargets();
+    for (const auto &ft : flushTargets) {
         vespalib::Executor::Task::UP task;
-        task = kv.second.getFlusher()->initFlush(currentSerial);
+        task = ft->initFlush(currentSerial);
         if (task.get() != NULL) {
             task->run();
         }
@@ -325,10 +345,14 @@ FlushableAttribute::SP
 AttributeManager::getFlushable(const vespalib::string &name)
 {
     auto wrap = findFlushable(name);
-    if (wrap != nullptr) {
-        return wrap->getFlusher();
-    }
-    return FlushableAttribute::SP();
+    return ((wrap != nullptr) ? wrap->getFlusher() : FlushableAttribute::SP());
+}
+
+AttributeManager::ShrinkerSP
+AttributeManager::getShrinker(const vespalib::string &name)
+{
+    auto wrap = findFlushable(name);
+    return ((wrap != nullptr) ? wrap->getShrinker() : ShrinkerSP());
 }
 
 size_t
@@ -452,6 +476,7 @@ AttributeManager::getFlushTargets() const
     list.reserve(_flushables.size());
     for (const auto &kv : _flushables) {
         list.push_back(kv.second.getFlusher());
+        list.push_back(kv.second.getShrinker());
     }
     return list;
 }
diff --git a/searchcore/src/vespa/searchcore/proton/attribute/attributemanager.h b/searchcore/src/vespa/searchcore/proton/attribute/attributemanager.h
index cb76a5a4aef..ca07a5e766f 100644
--- a/searchcore/src/vespa/searchcore/proton/attribute/attributemanager.h
+++ b/searchcore/src/vespa/searchcore/proton/attribute/attributemanager.h
@@ -98,7 +98,7 @@ private:
                                                      uint64_t serialNum,
                                                      const IAttributeFactory &factory);
 
-    void addAttribute(const AttributeWrap &attribute);
+    void addAttribute(const AttributeWrap &attribute, SerialNum shrinkSerialNum);
 
     AttributeVectorSP findAttribute(const vespalib::string &name) const;
 
@@ -149,6 +149,8 @@ public:
 
     FlushableAttributeSP getFlushable(const vespalib::string &name);
 
+    ShrinkerSP getShrinker(const vespalib::string &name);
+
     size_t getNumDocs() const;
 
     static void padAttribute(search::AttributeVector &v, uint32_t docIdLimit);
diff --git a/searchcore/src/vespa/searchcore/proton/attribute/flushableattribute.cpp b/searchcore/src/vespa/searchcore/proton/attribute/flushableattribute.cpp
index 260d819bec5..d4262fc0cb0 100644
--- a/searchcore/src/vespa/searchcore/proton/attribute/flushableattribute.cpp
+++ b/searchcore/src/vespa/searchcore/proton/attribute/flushableattribute.cpp
@@ -70,9 +70,6 @@ FlushableAttribute::Flusher::Flusher(FlushableAttribute & fattr, SerialNum syncT
     fattr._attr->commit(syncToken, syncToken);
     AttributeVector &attr = *_fattr._attr;
     // Called by attribute field writer executor
-    if (attr.canShrinkLidSpace()) {
-        attr.shrinkLidSpace();
-    }
     _flushFile = writer.getSnapshotDir(_syncToken) + "/" + attr.getName();
     attr.setBaseFileName(_flushFile);
     _saver = attr.initSave();
@@ -202,11 +199,7 @@ IFlushTarget::MemoryGain
 FlushableAttribute::getApproxMemoryGain() const
 {
     int64_t used(_attr->getStatus().getUsed());
-    int64_t canFree = _attr->getEstimatedShrinkLidSpaceGain();
-    if (canFree > used) {
-        canFree = used;
-    }
-    return MemoryGain(used, used - canFree);
+    return MemoryGain(used, used);
 }
 
 IFlushTarget::DiskGain