Add AttributeDirectory which manages the disk directory for a named

attribute vector.

11 files changed, 766 insertions, 6 deletions

diff --git a/searchcore/CMakeLists.txt b/searchcore/CMakeLists.txt
index 5634cde969e..7c96a567f55 100644
--- a/searchcore/CMakeLists.txt
+++ b/searchcore/CMakeLists.txt
@@ -61,6 +61,7 @@ vespa_define_module(
     src/tests/fdispatch/search_path
     src/tests/grouping
     src/tests/proton/attribute
+    src/tests/proton/attribute/attribute_directory
     src/tests/proton/attribute/attribute_manager
     src/tests/proton/attribute/attribute_populator
     src/tests/proton/attribute/attribute_usage_filter
diff --git a/searchcore/src/tests/proton/attribute/attribute_directory/CMakeLists.txt b/searchcore/src/tests/proton/attribute/attribute_directory/CMakeLists.txt
new file mode 100644
index 00000000000..3756220199e
--- /dev/null
+++ b/searchcore/src/tests/proton/attribute/attribute_directory/CMakeLists.txt
@@ -0,0 +1,8 @@
+# Copyright 2017 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+vespa_add_executable(searchcore_attribute_directory_test_app TEST
+    SOURCES
+    attribute_directory_test.cpp
+    DEPENDS
+    searchcore_attribute
+)
+vespa_add_test(NAME searchcore_attribute_directory_test_app COMMAND searchcore_attribute_directory_test_app)
diff --git a/searchcore/src/tests/proton/attribute/attribute_directory/DESC b/searchcore/src/tests/proton/attribute/attribute_directory/DESC
new file mode 100644
index 00000000000..f529e4ac5e4
--- /dev/null
+++ b/searchcore/src/tests/proton/attribute/attribute_directory/DESC
@@ -0,0 +1 @@
+AttributeDirectory test. Take a look at attribute_directory_test.cpp for details.
diff --git a/searchcore/src/tests/proton/attribute/attribute_directory/FILES b/searchcore/src/tests/proton/attribute/attribute_directory/FILES
new file mode 100644
index 00000000000..fccd4ba135a
--- /dev/null
+++ b/searchcore/src/tests/proton/attribute/attribute_directory/FILES
@@ -0,0 +1 @@
+attribute_directory_test.cpp
diff --git a/searchcore/src/tests/proton/attribute/attribute_directory/attribute_directory_test.cpp b/searchcore/src/tests/proton/attribute/attribute_directory/attribute_directory_test.cpp
new file mode 100644
index 00000000000..6c2a7734d42
--- /dev/null
+++ b/searchcore/src/tests/proton/attribute/attribute_directory/attribute_directory_test.cpp
@@ -0,0 +1,322 @@
+// Copyright 2017 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+#include <vespa/vespalib/testkit/testapp.h>
+#include <vespa/searchcore/proton/attribute/attribute_directory.h>
+#include <vespa/searchcore/proton/attribute/attributedisklayout.h>
+#include <vespa/searchcore/proton/test/directory_handler.h>
+#include <vespa/vespalib/stllike/string.h>
+#include <vespa/vespalib/stllike/asciistream.h>
+#include <vespa/vespalib/io/fileutil.h>
+#include <vespa/log/log.h>
+LOG_SETUP("attribute_directory_test");
+
+using search::IndexMetaInfo;
+using search::SerialNum;
+
+namespace proton {
+
+namespace {
+
+vespalib::string toString(IndexMetaInfo &info) {
+    vespalib::asciistream os;
+    bool first = true;
+    for (auto &snap : info.snapshots()) {
+        if (!first) {
+            os << ",";
+        }
+        first = false;
+        if (snap.valid) {
+            os << "v";
+        } else {
+            os << "i";
+        }
+        os << snap.syncToken;
+    }
+    return os.str();
+}
+
+bool gotAttributeDir(const std::shared_ptr<AttributeDirectory> &dir) {
+    return static_cast<bool>(dir);
+}
+
+bool gotWriter(const std::unique_ptr<AttributeDirectory::Writer> &writer) {
+    return static_cast<bool>(writer);
+}
+
+}
+
+struct Fixture : public test::DirectoryHandler
+{
+
+    std::shared_ptr<AttributeDiskLayout> _diskLayout;
+
+    Fixture()
+        : test::DirectoryHandler("attributes"),
+          _diskLayout(AttributeDiskLayout::create("attributes"))
+    {
+    }
+
+    ~Fixture() { }
+
+    vespalib::string getDir() { return _diskLayout->getBaseDir(); }
+
+    vespalib::string getAttrDir(const vespalib::string &name) { return getDir() + "/" + name; }
+
+    void assertAttributeDir(const vespalib::string &name) {
+        auto fileinfo = vespalib::stat(getAttrDir(name));
+        EXPECT_TRUE(static_cast<bool>(fileinfo));
+        EXPECT_TRUE(fileinfo->_directory);
+    }
+
+    void assertNotAttributeDir(const vespalib::string &name) {
+        auto fileinfo = vespalib::stat(getAttrDir(name));
+        EXPECT_FALSE(static_cast<bool>(fileinfo));
+    }
+
+    vespalib::string getSnapshotDirComponent(SerialNum serialNum) {
+        vespalib::asciistream os;
+        os << "snapshot-";
+        os << serialNum;
+        return os.str();
+    }
+
+    vespalib::string getSnapshotDir(const vespalib::string &name, SerialNum serialNum) {
+        return getAttrDir(name) + "/" + getSnapshotDirComponent(serialNum);
+    }
+
+    void assertSnapshotDir(const vespalib::string &name, SerialNum serialNum) {
+        vespalib::string snapDir = getSnapshotDir(name, serialNum);
+        auto fileinfo = vespalib::stat(snapDir);
+        EXPECT_TRUE(static_cast<bool>(fileinfo));
+        EXPECT_TRUE(fileinfo->_directory);
+    }
+
+    void assertNotSnapshotDir(const vespalib::string &name, SerialNum serialNum) {
+        vespalib::string snapDir = getSnapshotDir(name, serialNum);
+        auto fileinfo = vespalib::stat(snapDir);
+        EXPECT_FALSE(static_cast<bool>(fileinfo));
+    }
+
+    void assertSnapshots(const vespalib::string &name, const vespalib::string &exp) {
+        vespalib::string attrDir(getAttrDir(name));
+        IndexMetaInfo info(attrDir);
+        info.load();
+        vespalib::string act = toString(info);
+        EXPECT_EQUAL(exp, act);
+    }
+
+    auto createAttributeDir(const vespalib::string &name) { return _diskLayout->createAttributeDir(name); }
+    auto getAttributeDir(const vespalib::string &name) { return _diskLayout->getAttributeDir(name); }
+    void removeAttributeDir(const vespalib::string &name, SerialNum serialNum) { return _diskLayout->removeAttributeDir(name, serialNum); }
+    auto createFooAttrDir() { return createAttributeDir("foo"); }
+    auto getFooAttrDir() { return getAttributeDir("foo"); }
+    void removeFooAttrDir() { removeAttributeDir("foo", 10); }
+    void assertNotGetAttributeDir(const vespalib::string &name) {
+        auto dir = getAttributeDir(name);
+        EXPECT_FALSE(static_cast<bool>(dir));
+    }
+    void assertGetAttributeDir(const vespalib::string &name, std::shared_ptr<AttributeDirectory> expDir) {
+        auto dir = getAttributeDir(name);
+        EXPECT_TRUE(static_cast<bool>(dir));
+        EXPECT_EQUAL(expDir, dir);
+    }
+    void assertCreateAttributeDir(const vespalib::string &name, std::shared_ptr<AttributeDirectory> expDir) {
+        auto dir = getAttributeDir(name);
+        EXPECT_TRUE(static_cast<bool>(dir));
+        EXPECT_EQUAL(expDir, dir);
+    }
+
+    void setupFooSnapshots(SerialNum serialNum) {
+        auto dir = createFooAttrDir();
+        EXPECT_TRUE(gotAttributeDir(dir));
+        auto writer = dir->getWriter();
+        writer->createInvalidSnapshot(serialNum);
+        writer->markValidSnapshot(serialNum);
+        TEST_DO(assertAttributeDir("foo"));
+    }
+
+    void invalidateFooSnapshots(bool removeDir) {
+        auto dir = createFooAttrDir();
+        auto writer = dir->getWriter();
+        writer->invalidateOldSnapshots(10);
+        writer->removeInvalidSnapshots(removeDir);
+        TEST_DO(assertGetAttributeDir("foo", dir));
+    }
+
+
+    void testRemoveSnapshots(bool removeDir) {
+        TEST_DO(setupFooSnapshots(5));
+        TEST_DO(invalidateFooSnapshots(removeDir));
+        if (removeDir) {
+            TEST_DO(assertNotAttributeDir("foo"));
+        } else {
+            TEST_DO(assertAttributeDir("foo"));
+        }
+    }
+
+    void makeInvalidSnapshot(SerialNum serialNum) {
+        auto dir = createFooAttrDir();
+        EXPECT_TRUE(gotAttributeDir(dir));
+        dir->getWriter()->createInvalidSnapshot(serialNum);
+    }
+
+    void makeValidSnapshot(SerialNum serialNum) {
+        auto dir = createFooAttrDir();
+        auto writer = dir->getWriter();
+        writer->createInvalidSnapshot(serialNum);
+        writer->markValidSnapshot(serialNum);
+    }
+
+};
+
+TEST_F("Test that we can create attribute directory", Fixture)
+{
+    auto dir = f.createFooAttrDir();
+    EXPECT_TRUE(gotAttributeDir(dir));
+}
+
+
+TEST_F("Test that attribute directory is persistent", Fixture)
+{
+    TEST_DO(f.assertNotGetAttributeDir("foo"));
+    auto dir = f.createFooAttrDir();
+    EXPECT_TRUE(gotAttributeDir(dir));
+    TEST_DO(f.assertGetAttributeDir("foo", dir));
+}
+
+TEST_F("Test that we can remove attribute directory", Fixture)
+{
+    auto dir = f.createFooAttrDir();
+    EXPECT_TRUE(gotAttributeDir(dir));
+    TEST_DO(f.assertGetAttributeDir("foo", dir));
+    f.removeFooAttrDir();
+    TEST_DO(f.assertNotGetAttributeDir("foo"));
+}
+
+TEST_F("Test that we can create attribute directory with one snapshot", Fixture)
+{
+    TEST_DO(f.assertNotGetAttributeDir("foo"));
+    TEST_DO(f.assertNotAttributeDir("foo"));
+    auto dir = f.createFooAttrDir();
+    EXPECT_TRUE(gotAttributeDir(dir));
+    TEST_DO(f.assertNotAttributeDir("foo"));
+    dir->getWriter()->createInvalidSnapshot(1);
+    TEST_DO(f.assertAttributeDir("foo"));
+    TEST_DO(f.assertSnapshots("foo", "i1"));
+}
+
+TEST_F("Test that we can prune attribute snapshots", Fixture)
+{
+    auto dir = f.createFooAttrDir();
+    TEST_DO(f.assertNotAttributeDir("foo"));
+    auto writer = dir->getWriter();
+    writer->createInvalidSnapshot(2);
+    writer->markValidSnapshot(2);
+    writer->createInvalidSnapshot(4);
+    writer->markValidSnapshot(4);
+    writer.reset();
+    TEST_DO(f.assertAttributeDir("foo"));
+    TEST_DO(f.assertSnapshots("foo", "v2,v4"));
+    dir->getWriter()->invalidateOldSnapshots();
+    TEST_DO(f.assertSnapshots("foo", "i2,v4"));
+    dir->getWriter()->removeInvalidSnapshots(false);
+    TEST_DO(f.assertSnapshots("foo", "v4"));
+}
+
+TEST_F("Test that attribute directory is not removed if valid snapshots remain", Fixture)
+{
+    TEST_DO(f.setupFooSnapshots(20));
+    auto dir = f.getFooAttrDir();
+    EXPECT_TRUE(gotAttributeDir(dir));
+    dir->getWriter()->createInvalidSnapshot(30);
+    TEST_DO(f.assertSnapshots("foo", "v20,i30"));
+    TEST_DO(f.removeFooAttrDir());
+    TEST_DO(f.assertGetAttributeDir("foo", dir));
+    TEST_DO(f.assertAttributeDir("foo"));
+    TEST_DO(f.assertSnapshots("foo", "v20"));
+}
+
+TEST_F("Test that attribute directory is removed if no valid snapshots remain", Fixture)
+{
+    TEST_DO(f.setupFooSnapshots(5));
+    auto dir = f.getFooAttrDir();
+    EXPECT_TRUE(gotAttributeDir(dir));
+    dir->getWriter()->createInvalidSnapshot(30);
+    TEST_DO(f.assertSnapshots("foo", "v5,i30"));
+    TEST_DO(f.removeFooAttrDir());
+    TEST_DO(f.assertNotGetAttributeDir("foo"));
+    TEST_DO(f.assertNotAttributeDir("foo"));
+}
+
+TEST_F("Test that attribute directory is not removed due to pruning and disk dir is kept", Fixture)
+{
+    TEST_DO(f.testRemoveSnapshots(false));
+}
+
+TEST_F("Test that attribute directory is not removed due to pruning but disk dir is removed", Fixture)
+{
+    TEST_DO(f.testRemoveSnapshots(true));
+}
+
+TEST("Test that initial state tracks disk layout")
+{
+    vespalib::mkdir("attributes");
+    vespalib::mkdir("attributes/foo");
+    vespalib::mkdir("attributes/bar");
+    IndexMetaInfo fooInfo("attributes/foo");
+    IndexMetaInfo barInfo("attributes/bar");
+    fooInfo.addSnapshot({true, 4, "snapshot-4"});
+    fooInfo.addSnapshot({false, 8, "snapshot-8"});
+    fooInfo.save();
+    barInfo.addSnapshot({false, 5, "snapshot-5"});
+    barInfo.save();
+    Fixture f;
+    TEST_DO(f.assertAttributeDir("foo"));
+    TEST_DO(f.assertAttributeDir("bar"));
+    auto foodir = f.getFooAttrDir();
+    EXPECT_TRUE(gotAttributeDir(foodir));
+    auto bardir = f.getAttributeDir("bar");
+    EXPECT_TRUE(gotAttributeDir(bardir));
+    TEST_DO(f.assertNotGetAttributeDir("baz"));
+    TEST_DO(f.assertNotAttributeDir("baz"));
+    TEST_DO(f.assertSnapshots("foo", "v4,i8"));
+    TEST_DO(f.assertSnapshots("bar", "i5"));
+    f.makeInvalidSnapshot(12);
+    f.makeValidSnapshot(16);
+    TEST_DO(f.assertSnapshots("foo", "v4,i8,i12,v16"));
+}
+
+TEST_F("Test that snapshot removal removes correct snapshot directory", Fixture)
+{
+    TEST_DO(f.setupFooSnapshots(5));
+    vespalib::mkdir(f.getSnapshotDir("foo", 5));
+    vespalib::mkdir(f.getSnapshotDir("foo", 6));
+    TEST_DO(f.assertSnapshotDir("foo", 5));
+    TEST_DO(f.assertSnapshotDir("foo", 6));
+    TEST_DO(f.invalidateFooSnapshots(false));
+    TEST_DO(f.assertNotSnapshotDir("foo", 5));
+    TEST_DO(f.assertSnapshotDir("foo", 6));
+    TEST_DO(f.invalidateFooSnapshots(true));
+    TEST_DO(f.assertNotSnapshotDir("foo", 5));
+    TEST_DO(f.assertNotSnapshotDir("foo", 6));
+}
+
+TEST_F("Test that we can get nonblocking writer", Fixture)
+{
+    auto dir = f.createFooAttrDir();
+    auto writer = dir->getWriter();
+    EXPECT_TRUE(gotWriter(writer));
+    auto writer2 = dir->tryGetWriter();
+    EXPECT_FALSE(gotWriter(writer2));
+    writer.reset();
+    writer2 = dir->tryGetWriter();
+    EXPECT_TRUE(gotWriter(writer2));
+    writer = dir->tryGetWriter();
+    EXPECT_FALSE(gotWriter(writer));
+}
+
+}
+
+TEST_MAIN()
+{
+    TEST_RUN_ALL();
+}
diff --git a/searchcore/src/vespa/searchcore/proton/attribute/CMakeLists.txt b/searchcore/src/vespa/searchcore/proton/attribute/CMakeLists.txt
index a82fb1f3928..3972bd6dbc7 100644
--- a/searchcore/src/vespa/searchcore/proton/attribute/CMakeLists.txt
+++ b/searchcore/src/vespa/searchcore/proton/attribute/CMakeLists.txt
@@ -4,6 +4,7 @@ vespa_add_library(searchcore_attribute STATIC
     address_space_usage_stats.cpp
     attribute_collection_spec_factory.cpp
     attribute_collection_spec.cpp
+    attribute_directory.cpp
     attribute_factory.cpp
     attribute_initializer.cpp
     attribute_manager_explorer.cpp
diff --git a/searchcore/src/vespa/searchcore/proton/attribute/attribute_directory.cpp b/searchcore/src/vespa/searchcore/proton/attribute/attribute_directory.cpp
new file mode 100644
index 00000000000..3babd9556c7
--- /dev/null
+++ b/searchcore/src/vespa/searchcore/proton/attribute/attribute_directory.cpp
@@ -0,0 +1,251 @@
+// Copyright 2017 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include <vespa/fastos/fastos.h>
+#include "attribute_directory.h"
+#include "attributedisklayout.h"
+#include <vespa/searchlib/util/filekit.h>
+#include <vespa/vespalib/io/fileutil.h>
+#include <vespa/vespalib/stllike/asciistream.h>
+#include <vespa/log/log.h>
+LOG_SETUP(".proton.attribute.attribute_directory");
+
+using search::IndexMetaInfo;
+using search::SerialNum;
+
+namespace {
+
+vespalib::string
+getSnapshotDirComponent(uint64_t syncToken)
+{
+    vespalib::asciistream os;
+    os << "snapshot-" << syncToken;
+    return os.str();
+}
+
+}
+
+namespace proton {
+
+AttributeDirectory::AttributeDirectory(const std::shared_ptr<AttributeDiskLayout> &diskLayout,
+                                       const vespalib::string &name)
+    : _diskLayout(diskLayout),
+      _name(name),
+      _lastFlushTime(0),
+      _writer(nullptr),
+      _mutex(),
+      _cv(),
+      _snapInfo(getDirName())
+{
+    (void) _snapInfo.load();
+    SerialNum flushedSerialNum = getFlushedSerialNum();
+    if (flushedSerialNum != 0) {
+        vespalib::string dirName = getSnapshotDir(flushedSerialNum);
+        _lastFlushTime = search::FileKit::getModificationTime(dirName);
+    }
+}
+
+AttributeDirectory::~AttributeDirectory()
+{
+    std::lock_guard<std::mutex> guard(_mutex);
+    assert(_writer == nullptr);
+}
+
+vespalib::string
+AttributeDirectory::getDirName() const
+{
+    std::shared_ptr<AttributeDiskLayout> diskLayout;
+    {
+        std::unique_lock<std::mutex> guard(_mutex);
+        assert(!_diskLayout.expired());
+        diskLayout = _diskLayout.lock();
+    }
+    assert(diskLayout);
+    return AttributeDiskLayout::getAttributeBaseDir(diskLayout->getBaseDir(), _name);
+}
+
+SerialNum
+AttributeDirectory::getFlushedSerialNum() const
+{
+    std::lock_guard<std::mutex> guard(_mutex);
+    IndexMetaInfo::Snapshot bestSnap = _snapInfo.getBestSnapshot();
+    return bestSnap.valid ? bestSnap.syncToken : 0;
+}
+
+fastos::TimeStamp
+AttributeDirectory::getLastFlushTime() const
+{
+    return _lastFlushTime;
+}
+
+void
+AttributeDirectory::setLastFlushTime(fastos::TimeStamp lastFlushTime)
+{
+    _lastFlushTime = lastFlushTime;
+}
+
+void
+AttributeDirectory::saveSnapInfo()
+{
+    if (!_snapInfo.save()) {
+        vespalib::string dirName(getDirName());
+        LOG(warning,
+            "Could not save meta-info file for attribute vector '%s' to disk",
+            dirName.c_str());
+        abort();
+    }
+}
+
+vespalib::string
+AttributeDirectory::getSnapshotDir(search::SerialNum serialNum)
+{
+    auto snap = _snapInfo.getSnapshot(serialNum);
+    assert(snap.syncToken == serialNum);
+    vespalib::string dirName(getDirName());
+    return dirName + "/" + snap.dirName;
+}
+
+void
+AttributeDirectory::createInvalidSnapshot(SerialNum serialNum)
+{
+    IndexMetaInfo::Snapshot newSnap(false, serialNum, getSnapshotDirComponent(serialNum));
+    if (empty()) {
+        vespalib::string dirName(getDirName());
+        vespalib::mkdir(dirName, false);
+    }
+    {
+        std::lock_guard<std::mutex> guard(_mutex);
+        _snapInfo.addSnapshot(newSnap);
+    }
+    saveSnapInfo();
+}
+
+void
+AttributeDirectory::markValidSnapshot(SerialNum serialNum)
+{
+    {
+        std::lock_guard<std::mutex> guard(_mutex);
+        auto snap = _snapInfo.getSnapshot(serialNum);
+        assert(!snap.valid);
+        assert(snap.syncToken == serialNum);
+        _snapInfo.validateSnapshot(serialNum);
+    }
+    saveSnapInfo();
+}
+
+void
+AttributeDirectory::invalidateOldSnapshots(uint64_t serialNum)
+{
+    std::vector<SerialNum> toInvalidate;
+    {
+        std::lock_guard<std::mutex> guard(_mutex);
+        auto &list = _snapInfo.snapshots();
+        for (const auto &snap : list) {
+            if (snap.valid && snap.syncToken < serialNum) {
+                toInvalidate.emplace_back(snap.syncToken);
+            }
+        }
+        for (const auto &invalidSerialNum : toInvalidate) {
+            _snapInfo.invalidateSnapshot(invalidSerialNum);
+        }
+    }
+    if (!toInvalidate.empty()) {
+        saveSnapInfo();
+    }
+}
+
+void
+AttributeDirectory::invalidateOldSnapshots()
+{
+    auto best = _snapInfo.getBestSnapshot();
+    if (best.valid) {
+        invalidateOldSnapshots(best.syncToken);
+    }
+}
+
+bool
+AttributeDirectory::removeInvalidSnapshots(bool removeDir)
+{
+    std::vector<SerialNum> toRemove;
+    auto &list = _snapInfo.snapshots();
+    for (const auto &snap : list) {
+        if (!snap.valid) {
+            toRemove.emplace_back(snap.syncToken);
+        }
+    }
+    for (const auto &serialNum : toRemove) {
+        vespalib::string subDir(getSnapshotDir(serialNum));
+        vespalib::rmdir(subDir, true);
+    }
+    if (!toRemove.empty()) {
+        {
+            std::lock_guard<std::mutex> guard(_mutex);
+            for (const auto &serialNum : toRemove) {
+                _snapInfo.removeSnapshot(serialNum);
+            }
+        }
+        saveSnapInfo();
+    }
+    if (empty() && removeDir) {
+        vespalib::string dirName(getDirName());
+        vespalib::rmdir(dirName, true);
+        return true;
+    }
+    return false;
+}
+
+void
+AttributeDirectory::detach()
+{
+    assert(empty());
+    std::unique_lock<std::mutex> guard(_mutex);
+    _diskLayout.reset();
+}
+
+std::unique_ptr<AttributeDirectory::Writer>
+AttributeDirectory::getWriter()
+{
+    std::unique_lock<std::mutex> guard(_mutex);
+    while (_writer != nullptr) {
+        _cv.wait(guard);
+    }
+    std::shared_ptr<AttributeDiskLayout> diskLayout(_diskLayout.lock());
+    if (diskLayout) {
+        return std::make_unique<Writer>(*this);
+    } else {
+        return std::unique_ptr<Writer>(); // detached, no more writes
+    }
+}
+
+std::unique_ptr<AttributeDirectory::Writer>
+AttributeDirectory::tryGetWriter()
+{
+    std::lock_guard<std::mutex> guard(_mutex);
+    std::shared_ptr<AttributeDiskLayout> diskLayout(_diskLayout.lock());
+    if (diskLayout && _writer == nullptr) {
+        return std::make_unique<Writer>(*this);
+    } else {
+        return std::unique_ptr<Writer>();
+    }
+}
+
+bool
+AttributeDirectory::empty() const
+{
+    std::unique_lock<std::mutex> guard(_mutex);
+    return _snapInfo.snapshots().empty();
+}
+
+AttributeDirectory::Writer::Writer(AttributeDirectory &dir)
+    : _dir(dir)
+{
+    _dir._writer = this;
+}
+
+AttributeDirectory::Writer::~Writer()
+{
+    std::lock_guard<std::mutex> guard(_dir._mutex);
+    _dir._writer = nullptr;
+    _dir._cv.notify_all();
+}
+
+} // namespace proton
diff --git a/searchcore/src/vespa/searchcore/proton/attribute/attribute_directory.h b/searchcore/src/vespa/searchcore/proton/attribute/attribute_directory.h
new file mode 100644
index 00000000000..1a483da7daa
--- /dev/null
+++ b/searchcore/src/vespa/searchcore/proton/attribute/attribute_directory.h
@@ -0,0 +1,84 @@
+// Copyright 2017 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#pragma once
+
+#include <vespa/vespalib/stllike/string.h>
+#include <vespa/searchlib/common/indexmetainfo.h>
+#include <vespa/searchlib/common/serialnum.h>
+#include <memory>
+#include <mutex>
+#include <condition_variable>
+
+namespace proton {
+
+class AttributeDiskLayout;
+
+/*
+ * Class used to track changes to a directory containing saved
+ * snapshots of an attribute vector.
+ */
+class AttributeDirectory
+{
+public:
+    class Writer;
+    using SerialNum = search::SerialNum;
+
+private:
+    std::weak_ptr<AttributeDiskLayout> _diskLayout;
+    const vespalib::string _name;
+    fastos::TimeStamp      _lastFlushTime;
+    Writer                *_writer; // current writer
+    std::mutex             _flusherMutex;
+    mutable std::mutex     _mutex;
+    std::condition_variable _cv;
+    search::IndexMetaInfo  _snapInfo;
+
+    void saveSnapInfo();
+    vespalib::string getSnapshotDir(SerialNum serialNum);
+    void setLastFlushTime(fastos::TimeStamp lastFlushTime);
+    void createInvalidSnapshot(SerialNum serialNum);
+    void markValidSnapshot(SerialNum serialNum);
+    void invalidateOldSnapshots(SerialNum serialNum);
+    void invalidateOldSnapshots();
+    bool removeInvalidSnapshots(bool removeDir);
+    void detach();
+    vespalib::string getDirName() const;
+    bool empty() const;
+
+public:
+    AttributeDirectory(const std::shared_ptr<AttributeDiskLayout> &diskLayout,
+                       const vespalib::string &name);
+    ~AttributeDirectory();
+
+    /*
+     * Class to make changes to an attribute directory in a
+     * controlled manner. An exclusive lock is held during lifetime to
+     * ensure only one active writer at a time for an attribute directory.
+     */
+    class Writer {
+        AttributeDirectory &_dir;
+
+    public:
+        Writer(AttributeDirectory &dir);
+        ~Writer();
+
+        // methods called when saving an attribute.
+        void setLastFlushTime(fastos::TimeStamp lastFlushTime) { _dir.setLastFlushTime(lastFlushTime); }
+        void createInvalidSnapshot(SerialNum serialNum) { _dir.createInvalidSnapshot(serialNum); }
+        void markValidSnapshot(SerialNum serialNum) { _dir.markValidSnapshot(serialNum); }
+        vespalib::string getSnapshotDir(SerialNum serialNum) { return _dir.getSnapshotDir(serialNum); }
+
+        // methods called while pruning old snapshots or removing attribute
+        void invalidateOldSnapshots(SerialNum serialNum) { _dir.invalidateOldSnapshots(serialNum); }
+        void invalidateOldSnapshots() { _dir.invalidateOldSnapshots(); }
+        bool removeInvalidSnapshots(bool removeDir) { return _dir.removeInvalidSnapshots(removeDir); }
+        void detach() { _dir.detach(); }
+    };
+
+    std::unique_ptr<Writer> getWriter();
+    std::unique_ptr<Writer> tryGetWriter();
+    SerialNum getFlushedSerialNum() const;
+    fastos::TimeStamp getLastFlushTime() const;
+};
+
+} // namespace proton
diff --git a/searchcore/src/vespa/searchcore/proton/attribute/attributedisklayout.cpp b/searchcore/src/vespa/searchcore/proton/attribute/attributedisklayout.cpp
index 14865283a4b..f17f7c76566 100644
--- a/searchcore/src/vespa/searchcore/proton/attribute/attributedisklayout.cpp
+++ b/searchcore/src/vespa/searchcore/proton/attribute/attributedisklayout.cpp
@@ -6,6 +6,7 @@ LOG_SETUP(".proton.attribute.attributedisklayout");
 #include "attributedisklayout.h"
 #include <vespa/searchcommon/common/schemaconfigurer.h>
 #include <vespa/vespalib/io/fileutil.h>
+#include "attribute_directory.h"
 
 using search::IndexMetaInfo;
 using search::index::SchemaBuilder;
@@ -15,8 +16,10 @@ using search::AttributeVector;
 namespace proton
 {
 
-AttributeDiskLayout::AttributeDiskLayout(const vespalib::string &baseDir)
-    : _baseDir(baseDir)
+AttributeDiskLayout::AttributeDiskLayout(const vespalib::string &baseDir, PrivateConstructorTag)
+    : _baseDir(baseDir),
+      _mutex(),
+      _dirs()
 {
 }
 
@@ -207,5 +210,79 @@ AttributeDiskLayout::listAttributes(const vespalib::string &baseDir)
     return attributes;
 }
 
+void
+AttributeDiskLayout::scanDir()
+{
+    FastOS_DirectoryScan dir(_baseDir.c_str());
+    while (dir.ReadNext()) {
+        if (strcmp(dir.GetName(), "..") != 0 && strcmp(dir.GetName(), ".") != 0) {
+            if (dir.IsDirectory()) {
+                createAttributeDir(dir.GetName());
+            }
+        }
+    }
+}
+
+std::shared_ptr<AttributeDirectory>
+AttributeDiskLayout::getAttributeDir(const vespalib::string &name)
+{
+    std::shared_lock<std::shared_timed_mutex> guard(_mutex);
+    auto itr = _dirs.find(name);
+    if (itr == _dirs.end()) {
+        return std::shared_ptr<AttributeDirectory>();
+    } else {
+        return itr->second;
+    }
+}
+
+std::shared_ptr<AttributeDirectory>
+AttributeDiskLayout::createAttributeDir(const vespalib::string &name)
+{
+    std::unique_lock<std::shared_timed_mutex> guard(_mutex);
+    auto itr = _dirs.find(name);
+    if (itr == _dirs.end()) {
+        auto dir = std::make_shared<AttributeDirectory>(shared_from_this(), name);
+        auto insres = _dirs.insert(std::make_pair(name, dir));
+        assert(insres.second);
+        return dir;
+    } else {
+        return itr->second;
+    }
+}
+
+void
+AttributeDiskLayout::removeAttributeDir(const vespalib::string &name, search::SerialNum serialNum)
+{
+    auto dir = getAttributeDir(name);
+    if (dir) {
+        auto writer = dir->getWriter();
+        if (writer) {
+            writer->invalidateOldSnapshots(serialNum);
+            if (writer->removeInvalidSnapshots(true)) {
+                std::unique_lock<std::shared_timed_mutex> guard(_mutex);
+                auto itr = _dirs.find(name);
+                assert(itr != _dirs.end());
+                assert(dir.get() == itr->second.get());
+                _dirs.erase(itr);
+                writer->detach();
+            }
+        } else {
+            std::unique_lock<std::shared_timed_mutex> guard(_mutex);
+            auto itr = _dirs.find(name);
+            if (itr != _dirs.end()) {
+                assert(dir.get() != itr->second.get());
+            }
+        }
+    }
+}
+
+std::shared_ptr<AttributeDiskLayout>
+AttributeDiskLayout::create(const vespalib::string &baseDir)
+{
+    auto diskLayout = std::make_shared<AttributeDiskLayout>(baseDir, PrivateConstructorTag());
+    diskLayout->scanDir();
+    return diskLayout;
+}
+
 } // namespace proton
 
diff --git a/searchcore/src/vespa/searchcore/proton/attribute/attributedisklayout.h b/searchcore/src/vespa/searchcore/proton/attribute/attributedisklayout.h
index b084ae74ee7..466f9426f56 100644
--- a/searchcore/src/vespa/searchcore/proton/attribute/attributedisklayout.h
+++ b/searchcore/src/vespa/searchcore/proton/attribute/attributedisklayout.h
@@ -3,24 +3,34 @@
 #pragma once
 
 #include <vespa/searchlib/attribute/attributevector.h>
+#include <vespa/searchlib/common/serialnum.h>
 #include <vespa/searchlib/common/indexmetainfo.h>
 #include <vespa/vespalib/util/stringfmt.h>
 #include <vespa/searchcommon/common/schema.h>
+#include <mutex>
+#include <shared_mutex>
+#include <map>
 
 namespace proton {
 
+class AttributeDirectory;
 /**
  * Class with utility functions for handling the disk directory layout for attribute vectors.
  */
-class AttributeDiskLayout
+class AttributeDiskLayout : public std::enable_shared_from_this<AttributeDiskLayout>
 {
 private:
     const vespalib::string _baseDir;
+    mutable std::shared_timed_mutex _mutex;
+    std::map<vespalib::string, std::shared_ptr<AttributeDirectory>> _dirs;
+
     static vespalib::string getSnapshotDir(uint64_t syncToken);
     static vespalib::string getSnapshotRemoveDir(const vespalib::string &baseDir, const vespalib::string &snapDir);
 
+    void scanDir();
+    struct PrivateConstructorTag { };
 public:
-    explicit AttributeDiskLayout(const vespalib::string &baseDir);
+    explicit AttributeDiskLayout(const vespalib::string &baseDir, PrivateConstructorTag tag);
     ~AttributeDiskLayout();
     static vespalib::string  getAttributeBaseDir(const vespalib::string &baseDir, const vespalib::string &attrName);
     static search::AttributeVector::BaseName getAttributeFileName(const vespalib::string &baseDir, const vespalib::string &attrName, uint64_t syncToken);
@@ -29,6 +39,10 @@ public:
     static std::vector<vespalib::string> listAttributes(const vespalib::string &baseDir);
     const vespalib::string &getBaseDir() const { return _baseDir; }
     void createBaseDir();
+    std::shared_ptr<AttributeDirectory> getAttributeDir(const vespalib::string &name);
+    std::shared_ptr<AttributeDirectory> createAttributeDir(const vespalib::string &name);
+    void removeAttributeDir(const vespalib::string &name, search::SerialNum serialNum);
+    static std::shared_ptr<AttributeDiskLayout> create(const vespalib::string &baseDir);
 };
 
 } // namespace proton
diff --git a/searchcore/src/vespa/searchcore/proton/attribute/attributemanager.cpp b/searchcore/src/vespa/searchcore/proton/attribute/attributemanager.cpp
index 1cd5cbe2161..30bd4900f64 100644
--- a/searchcore/src/vespa/searchcore/proton/attribute/attributemanager.cpp
+++ b/searchcore/src/vespa/searchcore/proton/attribute/attributemanager.cpp
@@ -167,7 +167,7 @@ AttributeManager::AttributeManager(const vespalib::string &baseDir,
       _attributes(),
       _flushables(),
       _writableAttributes(),
-      _diskLayout(std::make_shared<AttributeDiskLayout>(baseDir)),
+      _diskLayout(AttributeDiskLayout::create(baseDir)),
       _documentSubDbName(documentSubDbName),
       _tuneFileAttributes(tuneFileAttributes),
       _fileHeaderContext(fileHeaderContext),
@@ -193,7 +193,7 @@ AttributeManager::AttributeManager(const vespalib::string &baseDir,
       _attributes(),
       _flushables(),
       _writableAttributes(),
-      _diskLayout(std::make_shared<AttributeDiskLayout>(baseDir)),
+      _diskLayout(AttributeDiskLayout::create(baseDir)),
       _documentSubDbName(documentSubDbName),
       _tuneFileAttributes(tuneFileAttributes),
       _fileHeaderContext(fileHeaderContext),