database/mariadb-columnstore-engine
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2026-10-23 21:49:05 +03:00
Code Activity

Merge pull request #3372 from denis0x0D/MCOL-5724_standalone

Browse Source 
MCOL-5724 Move journal to FDB.
This commit is contained in:
Leonid Fedorov
2025-05-23 04:15:20 +04:00
committed by GitHub
parent f877b17841 1071df26f2
commit 6923608169
10 changed files with 307 additions and 474 deletions
Download Patch File Download Diff File Expand all files Collapse all files

345
storage-manager/src/IOCoordinator.cpp
View File

@@ -16,6 +16,8 @@
MA 02110-1301, USA. */
#include "IOCoordinator.h"
#include "KVStorageInitializer.h"
#include "KVPrefixes.h"
#include "MetadataFile.h"
#include "Synchronizer.h"
#include <sys/types.h>
@@ -24,6 +26,7 @@
#include <unistd.h>
#include <stdlib.h>
#include <errno.h>
#include "cstring"
#include <boost/filesystem.hpp>
#define BOOST_SPIRIT_THREADSAFE
#include <boost/property_tree/json_parser.hpp>
@@ -189,7 +192,8 @@ ssize_t IOCoordinator::read(const char* _filename, uint8_t* data, off_t offset,
}
vector<metadataObject> relevants = meta.metadataRead(offset, length);
map<string, int> journalFDs, objectFDs;
map<string, int> objectFDs;
unordered_set<string> journalFDs;
map<string, string> keyToJournalName, keyToObjectName;
utils::VLArray<ScopedCloser> fdMinders(relevants.size() * 2);
int mindersIndex = 0;
@@ -211,29 +215,21 @@ ssize_t IOCoordinator::read(const char* _filename, uint8_t* data, off_t offset,
// later. not thinking about it for now.
// open all of the journal files that exist
string jFilename = (journalPath / firstDir / (key + ".journal")).string();
int fd = ::open(jFilename.c_str(), O_RDONLY);
if (fd >= 0)
const string journal = getJournalName((journalPath / firstDir / (key + ".journal")).string());
auto kvStorage = KVStorageInitializer::getStorageInstance();
auto tnx = kvStorage->createTransaction();
auto resultPairJournal = tnx->get(journal);
if (resultPairJournal.first)
{
keyToJournalName[key] = jFilename;
journalFDs[key] = fd;
fdMinders[mindersIndex++].fd = fd;
// fdMinders.push_back(SharedCloser(fd));
}
else if (errno != ENOENT)
{
int l_errno = errno;
fileLock.unlock();
cache->doneReading(firstDir, keys);
logger->log(LOG_CRIT, "IOCoordinator::read(): Got an unexpected error opening %s, error was '%s'",
jFilename.c_str(), strerror_r(l_errno, buf, 80));
errno = l_errno;
return -1;
keyToJournalName[key] = journal;
journalFDs.insert(key);
}
// else
// Journal already merged with object file.
// open all of the objects
string oFilename = (cachePath / firstDir / key).string();
fd = ::open(oFilename.c_str(), O_RDONLY);
auto fd = ::open(oFilename.c_str(), O_RDONLY);
if (fd < 0)
{
int l_errno = errno;
@@ -247,9 +243,7 @@ ssize_t IOCoordinator::read(const char* _filename, uint8_t* data, off_t offset,
keyToObjectName[key] = oFilename;
objectFDs[key] = fd;
fdMinders[mindersIndex++].fd = fd;
// fdMinders.push_back(SharedCloser(fd));
}
// fileLock.unlock();
// ^^^ TODO: The original version unlocked the file at the above position. On second glance,
// I'm not sure how we can guarantee the journal files won't be modified during the loads below.
@@ -857,7 +851,14 @@ int IOCoordinator::_truncate(const bf::path& bfpath, size_t newSize, ScopedFileL
if (result & 0x1)
replicator->remove(cachePath / firstDir / objects[i].key);
if (result & 0x2)
replicator->remove(journalPath / firstDir / (objects[i].key + ".journal"));
{
const auto journalName =
getJournalName((journalPath / firstDir / (objects[i].key + ".journal")).string());
const auto journalSizeName =
getJournalName((journalPath / firstDir / (objects[i].key + "_size.journal")).string());
replicator->removeJournal(journalName);
replicator->removeJournalSize(journalSizeName);
}
deletedObjects.push_back(objects[i].key);
}
if (!deletedObjects.empty())
@@ -905,7 +906,11 @@ void IOCoordinator::deleteMetaFile(const bf::path& file)
if (result & 0x2)
{
++iocFilesDeleted;
replicator->remove(journalPath / firstDir / (object.key + ".journal"));
const auto journalName = getJournalName((journalPath / firstDir / (object.key + ".journal")).string());
const auto journalSizeName =
getJournalName((journalPath / firstDir / (object.key + "_size.journal")).string());
replicator->removeJournalSize(journalSizeName);
replicator->removeJournal(journalName);
}
deletedObjects.push_back(object.key);
}
@@ -1021,20 +1026,6 @@ int IOCoordinator::copyFile(const char* _filename1, const char* _filename2)
int err;
char errbuf[80];
// since we don't implement mkdir(), assume the caller did that and
// create any necessary parent dirs for filename2
try
{
bf::create_directories(metaFile2.parent_path());
}
catch (bf::filesystem_error& e)
{
logger->log(LOG_CRIT, "IOCoordinator::copyFile(): failed to create directory %s. Got %s",
metaFile2.parent_path().string().c_str(), strerror_r(e.code().value(), errbuf, 80));
errno = e.code().value();
return -1;
}
vector<pair<string, size_t>> newJournalEntries;
ScopedReadLock lock(this, filename1);
ScopedWriteLock lock2(this, filename2);
@@ -1060,8 +1051,6 @@ int IOCoordinator::copyFile(const char* _filename1, const char* _filename2)
{
for (const auto& object : objects)
{
bf::path journalFile = journalPath / firstDir1 / (object.key + ".journal");
// originalLength = the length of the object before journal entries.
// the length in the metadata is the length after journal entries
size_t originalLength = MetadataFile::getLengthFromKey(object.key);
@@ -1095,27 +1084,36 @@ int IOCoordinator::copyFile(const char* _filename1, const char* _filename2)
object.key + ": " + strerror_r(errno, errbuf, 80));
}
// if there's a journal file for this object, make a copy
if (bf::exists(journalFile))
const auto journalName = getJournalName((journalPath / firstDir1 / (object.key + ".journal")).string());
auto keyGen = std::make_shared<FDBCS::BoostUIDKeyGenerator>();
FDBCS::BlobHandler journalReader(keyGen);
auto kvStorage = KVStorageInitializer::getStorageInstance();
auto resultPair = journalReader.readBlob(kvStorage, journalName);
if (resultPair.first)
{
bf::path newJournalFile = journalPath / firstDir2 / (newObj.key + ".journal");
try
const auto& journalData = resultPair.second;
const std::string oldJournalDataHeader = resultPair.second;
const auto newJournalName =
getJournalName((journalPath / firstDir2 / (newObj.key + ".journal")).string());
const auto newJournalSizeName =
getJournalName((journalPath / firstDir2 / (newObj.key + "_size.journal")).string());
FDBCS::BlobHandler journalWriter(keyGen);
if (!journalWriter.writeBlob(kvStorage, newJournalName, journalData))
logger->log(LOG_CRIT, "Cannot write new journal, while copying files.");
{
bf::copy_file(journalFile, newJournalFile);
size_t tmp = bf::file_size(newJournalFile);
++iocJournalsCreated;
iocBytesRead += tmp;
iocBytesWritten += tmp;
cache->newJournalEntry(firstDir2, tmp);
newJournalEntries.push_back(pair<string, size_t>(newObj.key, tmp));
}
catch (bf::filesystem_error& e)
{
throw CFException(e.code().value(), string("IOCoordinator::copyFile(): source = ") + filename1 +
", dest = " + filename2 + ". Got an error copying " +
journalFile.string() + ": " +
strerror_r(e.code().value(), errbuf, 80));
auto tnx = kvStorage->createTransaction();
tnx->set(newJournalSizeName, to_string(journalData.size()));
if (!tnx->commit())
logger->log(LOG_CRIT, "Cannot write new journal size, while copying files.");
}
++iocJournalsCreated;
iocBytesRead += journalData.size();
iocBytesWritten += journalData.size();
cache->newJournalEntry(firstDir2, journalData.size());
newJournalEntries.push_back(pair<string, size_t>(newObj.key, journalData.size()));
}
}
}
@@ -1126,9 +1124,12 @@ int IOCoordinator::copyFile(const char* _filename1, const char* _filename2)
cs->deleteObject(newObject.key);
for (auto& jEntry : newJournalEntries)
{
bf::path fullJournalPath = journalPath / firstDir2 / (jEntry.first + ".journal");
const auto fullJournalPath =
getJournalName((journalPath / firstDir2 / (jEntry.first + ".journal")).string());
auto kvStorage = KVStorageInitializer::getStorageInstance();
auto tnx = kvStorage->createTransaction();
tnx->remove(fullJournalPath);
cache->deletedJournal(firstDir2, bf::file_size(fullJournalPath));
bf::remove(fullJournalPath);
}
errno = e.l_errno;
return -1;
@@ -1163,28 +1164,21 @@ const bf::path& IOCoordinator::getMetadataPath() const
// first byte after the header.
// update: had to make it also return the header; the boost json parser does not stop at either
// a null char or the end of an object.
std::shared_ptr<char[]> seekToEndOfHeader1(int fd, size_t* _bytesRead)
std::shared_ptr<char[]> seekToEndOfHeader1_(const std::string& dataStr, size_t* _bytesRead)
{
//::lseek(fd, 0, SEEK_SET);
std::shared_ptr<char[]> ret(new char[100]);
int err;
const size_t numBytesToRead = dataStr.size();
std::shared_ptr<char[]> ret(new char[numBytesToRead]);
std::memcpy(ret.get(), &dataStr[0], numBytesToRead);
err = ::read(fd, ret.get(), 100);
if (err < 0)
{
char buf[80];
throw runtime_error("seekToEndOfHeader1 got: " + string(strerror_r(errno, buf, 80)));
}
for (int i = 0; i < err; i++)
for (uint32_t i = 0; i < numBytesToRead; i++)
{
if (ret[i] == 0)
{
::lseek(fd, i + 1, SEEK_SET);
*_bytesRead = i + 1;
return ret;
}
}
throw runtime_error("seekToEndOfHeader1: did not find the end of the header");
throw runtime_error("seekToEndOfHeader1_: did not find the end of the header");
}
int IOCoordinator::mergeJournal(int objFD, int journalFD, uint8_t* buf, off_t offset, size_t* len) const
@@ -1195,7 +1189,7 @@ int IOCoordinator::mergeJournal(int objFD, int journalFD, uint8_t* buf, off_t of
std::shared_ptr<uint8_t[]> IOCoordinator::mergeJournal(const char* object, const char* journal, off_t offset,
size_t len, size_t* _bytesReadOut) const
{
int objFD, journalFD;
int objFD;
std::shared_ptr<uint8_t[]> ret;
size_t l_bytesRead = 0;
@@ -1208,7 +1202,6 @@ std::shared_ptr<uint8_t[]> IOCoordinator::mergeJournal(const char* object, const
ScopedCloser s1(objFD);
ret.reset(new uint8_t[len]);
// read the object into memory
size_t count = 0;
if (offset != 0)
@@ -1249,29 +1242,34 @@ std::shared_ptr<uint8_t[]> IOCoordinator::mergeJournal(const char* object, const
return ret;
}
journalFD = ::open(journal, O_RDONLY);
if (journalFD < 0)
// Read journal.
auto keyGen = std::make_shared<FDBCS::BoostUIDKeyGenerator>();
FDBCS::BlobHandler journalReader(keyGen);
auto kvStorage = KVStorageInitializer::getStorageInstance();
auto resultPairJournal = journalReader.readBlob(kvStorage, journal);
if (!resultPairJournal.first)
{
*_bytesReadOut = l_bytesRead;
return ret;
}
ScopedCloser s2(journalFD);
std::shared_ptr<char[]> headertxt = seekToEndOfHeader1(journalFD, &l_bytesRead);
const std::string& journalData = resultPairJournal.second;
size_t journalOffset = 0;
std::shared_ptr<char[]> headertxt = seekToEndOfHeader1_(journalData, &journalOffset);
stringstream ss;
ss << headertxt.get();
boost::property_tree::ptree header;
boost::property_tree::json_parser::read_json(ss, header);
assert(header.get<int>("version") == 1);
l_bytesRead += journalOffset;
const size_t journalSize = journalData.size();
// start processing the entries
while (1)
while (journalOffset < journalSize)
{
uint64_t offlen[2];
int err = ::read(journalFD, &offlen, 16);
if (err == 0) // got EOF
break;
assert(err == 16);
std::memcpy(&offlen, &journalData[journalOffset], 16);
journalOffset += 16;
l_bytesRead += 16;
// if this entry overlaps, read the overlapping section
@@ -1284,45 +1282,15 @@ std::shared_ptr<uint8_t[]> IOCoordinator::mergeJournal(const char* object, const
// seek to the portion of the entry to start reading at
if (startReadingAt != offlen[0])
::lseek(journalFD, startReadingAt - offlen[0], SEEK_CUR);
uint count = 0;
while (count < lengthOfRead)
{
err = ::read(journalFD, &ret[startReadingAt - offset + count], lengthOfRead - count);
if (err < 0)
{
int l_errno = errno;
char buf[80];
logger->log(LOG_ERR, "mergeJournal: got %s", strerror_r(l_errno, buf, 80));
ret.reset();
errno = l_errno;
l_bytesRead += count;
goto out;
}
else if (err == 0)
{
logger->log(LOG_ERR,
"mergeJournal: got early EOF. offset=%ld, len=%ld, jOffset=%ld, jLen=%ld,"
" startReadingAt=%ld, lengthOfRead=%ld",
offset, len, offlen[0], offlen[1], startReadingAt, lengthOfRead);
ret.reset();
l_bytesRead += count;
goto out;
}
count += err;
}
l_bytesRead += lengthOfRead;
// advance the file pos if we didn't read to the end of the entry
if (startReadingAt - offlen[0] + lengthOfRead != offlen[1])
::lseek(journalFD, offlen[1] - (lengthOfRead + startReadingAt - offlen[0]), SEEK_CUR);
journalOffset += startReadingAt - offlen[0];
//::lseek(journalFD, startReadingAt - offlen[0], SEEK_CUR);
std::memcpy(&ret[startReadingAt - offset], &journalData[journalOffset], lengthOfRead);
journalOffset += lengthOfRead;
}
else
// skip over this journal entry
::lseek(journalFD, offlen[1], SEEK_CUR);
journalOffset += offlen[1];
}
out:
*_bytesReadOut = l_bytesRead;
return ret;
}
@@ -1332,19 +1300,18 @@ out:
int IOCoordinator::mergeJournalInMem(std::shared_ptr<uint8_t[]>& objData, size_t len, const char* journalPath,
size_t* _bytesReadOut) const
{
// if the journal is over some size threshold (100MB for now why not),
// use the original low-mem-usage version
if (len > (100 << 20))
return mergeJournalInMem_bigJ(objData, len, journalPath, _bytesReadOut);
size_t l_bytesRead = 0;
int journalFD = ::open(journalPath, O_RDONLY);
if (journalFD < 0)
auto kvStorage = KVStorageInitializer::getStorageInstance();
auto keyGen = std::make_shared<FDBCS::BoostUIDKeyGenerator>();
FDBCS::BlobHandler blobReader(keyGen);
auto resultPair = blobReader.readBlob(kvStorage, journalPath);
if (!resultPair.first)
return -1;
ScopedCloser s(journalFD);
const std::string& journalData = resultPair.second;
size_t journalOffset = 0;
// grab the journal header and make sure the version is 1
std::shared_ptr<char[]> headertxt = seekToEndOfHeader1(journalFD, &l_bytesRead);
std::shared_ptr<char[]> headertxt = seekToEndOfHeader1_(journalData, &l_bytesRead);
stringstream ss;
ss << headertxt.get();
boost::property_tree::ptree header;
@@ -1352,136 +1319,46 @@ int IOCoordinator::mergeJournalInMem(std::shared_ptr<uint8_t[]>& objData, size_t
assert(header.get<int>("version") == 1);
// read the journal file into memory
size_t journalBytes = ::lseek(journalFD, 0, SEEK_END) - l_bytesRead;
::lseek(journalFD, l_bytesRead, SEEK_SET);
boost::scoped_array<uint8_t> journalData(new uint8_t[journalBytes]);
size_t journalBytes = journalData.size() - l_bytesRead;
journalOffset += l_bytesRead;
size_t readCount = 0;
while (readCount < journalBytes)
{
ssize_t err = ::read(journalFD, &journalData[readCount], journalBytes - readCount);
if (err < 0)
{
char buf[80];
int l_errno = errno;
logger->log(LOG_ERR, "mergeJournalInMem: got %s", strerror_r(errno, buf, 80));
errno = l_errno;
return -1;
}
else if (err == 0)
{
logger->log(LOG_ERR, "mergeJournalInMem: got early EOF");
errno = ENODATA; // is there a better errno for early EOF?
return -1;
}
readCount += err;
l_bytesRead += err;
}
readCount += journalBytes;
l_bytesRead += journalBytes;
const size_t journalSize = journalData.size();
// start processing the entries
size_t offset = 0;
while (offset < journalBytes)
while (journalOffset < journalSize)
{
if (offset + 16 >= journalBytes)
if (journalOffset + 16 >= journalSize)
{
logger->log(LOG_ERR, "mergeJournalInMem: got early EOF");
errno = ENODATA; // is there a better errno for early EOF?
return -1;
}
uint64_t* offlen = (uint64_t*)&journalData[offset];
offset += 16;
uint64_t* offlen = (uint64_t*)&journalData[journalOffset];
journalOffset += 16;
uint64_t startReadingAt = offlen[0];
uint64_t lengthOfRead = offlen[1];
if (startReadingAt > len)
{
offset += offlen[1];
continue;
}
if (startReadingAt + lengthOfRead > len)
lengthOfRead = len - startReadingAt;
if (offset + lengthOfRead > journalBytes)
{
logger->log(LOG_ERR, "mergeJournalInMem: got early EOF");
errno = ENODATA; // is there a better errno for early EOF?
return -1;
}
memcpy(&objData[startReadingAt], &journalData[offset], lengthOfRead);
offset += offlen[1];
}
*_bytesReadOut = l_bytesRead;
return 0;
}
int IOCoordinator::mergeJournalInMem_bigJ(std::shared_ptr<uint8_t[]>& objData, size_t len,
const char* journalPath, size_t* _bytesReadOut) const
{
size_t l_bytesRead = 0;
int journalFD = ::open(journalPath, O_RDONLY);
if (journalFD < 0)
return -1;
ScopedCloser s(journalFD);
// grab the journal header and make sure the version is 1
std::shared_ptr<char[]> headertxt = seekToEndOfHeader1(journalFD, &l_bytesRead);
stringstream ss;
ss << headertxt.get();
boost::property_tree::ptree header;
boost::property_tree::json_parser::read_json(ss, header);
assert(header.get<int>("version") == 1);
// start processing the entries
while (1)
{
uint64_t offlen[2];
int err = ::read(journalFD, &offlen, 16);
if (err == 0) // got EOF
if (lengthOfRead == 0)
break;
else if (err < 16)
{
// punting on this
cout << "mergeJournalInMem: failed to read a journal entry header in one attempt. fixme..." << endl;
errno = ENODATA;
return -1;
}
l_bytesRead += 16;
uint64_t startReadingAt = offlen[0];
uint64_t lengthOfRead = offlen[1];
if (startReadingAt > len)
{
::lseek(journalFD, offlen[1], SEEK_CUR);
journalOffset += offlen[1];
continue;
}
if (startReadingAt + lengthOfRead > len)
lengthOfRead = len - startReadingAt;
uint count = 0;
while (count < lengthOfRead)
if (journalOffset + lengthOfRead > journalSize)
{
err = ::read(journalFD, &objData[startReadingAt + count], lengthOfRead - count);
if (err < 0)
{
char buf[80];
int l_errno = errno;
logger->log(LOG_ERR, "mergeJournalInMem: got %s", strerror_r(errno, buf, 80));
errno = l_errno;
return -1;
}
else if (err == 0)
{
logger->log(LOG_ERR, "mergeJournalInMem: got early EOF");
errno = ENODATA; // is there a better errno for early EOF?
return -1;
}
count += err;
logger->log(LOG_ERR, "mergeJournalInMem: got early EOF");
errno = ENODATA; // is there a better errno for early EOF?
return -1;
}
l_bytesRead += lengthOfRead;
if (lengthOfRead < offlen[1])
::lseek(journalFD, offlen[1] - lengthOfRead, SEEK_CUR);
std::memcpy(&objData[startReadingAt], &journalData[journalOffset], lengthOfRead);
journalOffset += offlen[1];
}
*_bytesReadOut = l_bytesRead;
return 0;

9
storage-manager/src/IOCoordinator.h
View File

@@ -37,7 +37,7 @@
namespace storagemanager
{
std::shared_ptr<char[]> seekToEndOfHeader1(int fd, size_t* bytesRead);
std::shared_ptr<char[]> seekToEndOfHeader1_(const std::string& dataStr, size_t* bytesRead);
class IOCoordinator : public boost::noncopyable
{
@@ -59,15 +59,10 @@ class IOCoordinator : public boost::noncopyable
// len should be set to the length of the data requested
std::shared_ptr<uint8_t[]> mergeJournal(const char* objectPath, const char* journalPath, off_t offset,
size_t len, size_t* sizeRead) const;
// this version modifies object data in memory, given the journal filename. Processes the whole object
// and whole journal file.
int mergeJournalInMem(std::shared_ptr<uint8_t[]>& objData, size_t len, const char* journalPath,
size_t* sizeRead) const;
// this version of MJIM has a higher IOPS requirement and lower mem usage.
int mergeJournalInMem_bigJ(std::shared_ptr<uint8_t[]>& objData, size_t len, const char* journalPath,
size_t* sizeRead) const;
size_t* sizeRead) const;
// this version takes already-open file descriptors, and an already-allocated buffer as input.
// file descriptor are positioned, eh, best not to assume anything about their positions

9
storage-manager/src/KVPrefixes.cpp
View File

@@ -15,10 +15,15 @@
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
MA 02110-1301, USA. */
#include "KVPrefixes.hpp"
#include "KVPrefixes.h"
namespace storagemanager
{
// FDB recommends keep the key size up to 32 bytes.
const char* KVPrefixes[2] = {/*OWNERSHIP*/ "SM_O_", /*META*/ "SM_M_"};
const char* KVPrefixes[3] = {/*OWNERSHIP*/ "SM_O_", /*META*/ "SM_M_", /*JOURNAL*/ "SM_J_"};
std::string getJournalName(const std::string& key)
{
return KVPrefixes[static_cast<uint32_t>(KVPrefixId::SM_JOURNAL)] + key;
}
} // namespace storagemanager

7
storage-manager/src/KVPrefixes.hpp → storage-manager/src/KVPrefixes.h
View File

@@ -16,14 +16,19 @@
MA 02110-1301, USA. */
#pragma once
#include <string>
#include <cstdint>
namespace storagemanager
{
enum class KVPrefixId
{
SM_OWNERSHIP = 0,
SM_META
SM_META,
SM_JOURNAL
};
extern const char* KVPrefixes[];
std::string getJournalName(const std::string &key);
} // namespace storagemanager

2
storage-manager/src/MetadataFile.cpp
View File

@@ -45,7 +45,7 @@
#include <boost/uuid/random_generator.hpp>
#include <unistd.h>
#include "fdbcs.hpp"
#include "KVPrefixes.hpp"
#include "KVPrefixes.h"
#define max(x, y) (x > y ? x : y)
#define min(x, y) (x < y ? x : y)

2
storage-manager/src/Ownership.cpp
View File

@@ -20,7 +20,7 @@
#include "Cache.h"
#include "Synchronizer.h"
#include "KVStorageInitializer.h"
#include "KVPrefixes.hpp"
#include "KVPrefixes.h"
#include "fdbcs.hpp"
#include <sys/types.h>
#include <sys/stat.h>

21
storage-manager/src/PrefixCache.cpp
View File

@@ -17,6 +17,8 @@
#include "PrefixCache.h"
#include "Cache.h"
#include "KVPrefixes.h"
#include "KVStorageInitializer.h"
#include "Config.h"
#include "Downloader.h"
#include "Synchronizer.h"
@@ -575,7 +577,8 @@ void PrefixCache::rename(const string& oldKey, const string& newKey, ssize_t siz
int PrefixCache::ifExistsThenDelete(const string& key)
{
bf::path cachedPath = cachePrefix / key;
bf::path journalPath = journalPrefix / (key + ".journal");
const auto journalName = getJournalName((journalPrefix / (key + ".journal")).string());
const auto journalSizeName = getJournalName((journalPrefix / (key + "_size.journal")).string());
boost::unique_lock<boost::mutex> s(lru_mutex);
bool objectExists = false;
@@ -593,16 +596,18 @@ int PrefixCache::ifExistsThenDelete(const string& key)
else // let makeSpace() delete it if it's already in progress
return 0;
}
bool journalExists = bf::exists(journalPath);
// assert(objectExists == bf::exists(cachedPath));
auto kvStorage = KVStorageInitializer::getStorageInstance();
auto tnx = kvStorage->createTransaction();
auto journalResult = tnx->get(journalName);
auto journalSizeResult = tnx->get(journalSizeName);
bool journalExists = journalSizeResult.first;
size_t objectSize = (objectExists ? bf::file_size(cachedPath) : 0);
// size_t objectSize = (objectExists ? MetadataFile::getLengthFromKey(key) : 0);
size_t journalSize = (journalExists ? bf::file_size(journalPath) : 0);
size_t journalSize = 0;
if (journalExists)
journalSize = std::atoi(journalSizeResult.second.c_str());
currentCacheSize -= (objectSize + journalSize);
// assert(!objectExists || objectSize == bf::file_size(cachedPath));
return (objectExists ? 1 : 0) | (journalExists ? 2 : 0);
}

269
storage-manager/src/Replicator.cpp
View File

@@ -20,6 +20,9 @@
#include "SMLogging.h"
#include "Utilities.h"
#include "Cache.h"
#include "KVStorageInitializer.h"
#include "KVPrefixes.h"
#include "fdbcs.hpp"
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
@@ -121,9 +124,8 @@ void Replicator::printKPIs() const
int Replicator::newObject(const boost::filesystem::path& filename, const uint8_t* data, off_t offset,
size_t length)
{
const string objectFilename = msCachePath + "/" + filename.string();
int fd, err;
string objectFilename = msCachePath + "/" + filename.string();
OPEN(objectFilename.c_str(), O_WRONLY | O_CREAT);
size_t count = 0;
while (count < length)
@@ -215,54 +217,48 @@ ssize_t Replicator::_write(int fd, const void* data, size_t length)
Benefits would be data integrity, and possibly add'l parallelism. The downside is of course, a higher
number of IO ops for the same operation.
*/
int Replicator::addJournalEntry(const boost::filesystem::path& filename, const uint8_t* data, off_t offset,
size_t length)
{
int fd, err;
uint64_t offlen[] = {(uint64_t)offset, length};
size_t count = 0;
int version = 1;
int l_errno;
char errbuf[80];
bool bHeaderChanged = false;
string headerRollback = "";
string journalFilename = msJournalPath + "/" + filename.string() + ".journal";
const int version = 1;
const auto journalName = getJournalName(msJournalPath + "/" + filename.string() + ".journal");
const auto journalSizeName = getJournalName(msJournalPath + "/" + filename.string() + "_size" + ".journal");
boost::filesystem::path firstDir = *((filename).begin());
uint64_t thisEntryMaxOffset = (offset + length - 1);
const uint64_t thisEntryMaxOffset = (offset + length - 1);
string dataStr;
size_t dataStrOffset = 0;
uint64_t currentMaxOffset = 0;
bool exists = boost::filesystem::exists(journalFilename);
OPEN(journalFilename.c_str(), (exists ? O_RDWR : O_WRONLY | O_CREAT))
if (!exists)
auto kvStorage = KVStorageInitializer::getStorageInstance();
auto keyGen = std::make_shared<FDBCS::BoostUIDKeyGenerator>();
FDBCS::BlobHandler journalHandler(keyGen);
auto resultPair = journalHandler.readBlob(kvStorage, journalName);
const std::string& journalData = resultPair.second;
const bool journalExists = resultPair.first;
if (!journalExists)
{
bHeaderChanged = true;
// create new journal file with header
string header = (boost::format("{ \"version\" : \"%03i\", \"max_offset\" : \"%011u\" }") % version %
thisEntryMaxOffset)
.str();
err = _write(fd, header.c_str(), header.length() + 1);
l_errno = errno;
repHeaderDataWritten += (header.length() + 1);
if ((uint)err != (header.length() + 1))
{
// return the error because the header for this entry on a new journal file failed
mpLogger->log(LOG_CRIT, "Replicator::addJournalEntry: Writing journal header failed (%s).",
strerror_r(l_errno, errbuf, 80));
errno = l_errno;
return err;
}
Cache::get()->newJournalEntry(firstDir, header.length() + 1);
const size_t headerLength = header.length();
dataStr.resize(headerLength + 1 + JOURNAL_ENTRY_HEADER_SIZE + length);
std::memcpy(&dataStr[dataStrOffset], header.c_str(), headerLength);
// Specifies the end of the header.
dataStr[headerLength] = 0;
dataStrOffset = headerLength + 1;
repHeaderDataWritten += headerLength + 1;
Cache::get()->newJournalEntry(firstDir, headerLength + 1);
++replicatorJournalsCreated;
}
else
{
// read the existing header and check if max_offset needs to be updated
size_t tmp;
std::shared_ptr<char[]> headertxt;
try
{
headertxt = seekToEndOfHeader1(fd, &tmp);
headertxt = seekToEndOfHeader1_(journalData, &tmp);
}
catch (std::runtime_error& e)
{
@@ -278,7 +274,6 @@ int Replicator::addJournalEntry(const boost::filesystem::path& filename, const u
}
stringstream ss;
ss << headertxt.get();
headerRollback = headertxt.get();
boost::property_tree::ptree header;
try
{
@@ -297,164 +292,61 @@ int Replicator::addJournalEntry(const boost::filesystem::path& filename, const u
return -1;
}
assert(header.get<int>("version") == 1);
uint64_t currentMaxOffset = header.get<uint64_t>("max_offset");
const uint64_t currentMaxOffset = header.get<uint64_t>("max_offset");
dataStr.resize(journalData.size() + JOURNAL_ENTRY_HEADER_SIZE + length);
size_t journalOffset = 0;
if (thisEntryMaxOffset > currentMaxOffset)
{
bHeaderChanged = true;
string header = (boost::format("{ \"version\" : \"%03i\", \"max_offset\" : \"%011u\" }") % version %
thisEntryMaxOffset)
.str();
err = _pwrite(fd, header.c_str(), header.length() + 1, 0);
l_errno = errno;
repHeaderDataWritten += (header.length() + 1);
if ((uint)err != (header.length() + 1))
{
// only the header was possibly changed rollback attempt
mpLogger->log(LOG_CRIT,
"Replicator::addJournalEntry: Updating journal header failed. "
"Attempting to rollback and continue.");
int rollbackErr = _pwrite(fd, headerRollback.c_str(), headerRollback.length() + 1, 0);
if ((uint)rollbackErr == (headerRollback.length() + 1))
mpLogger->log(LOG_CRIT, "Replicator::addJournalEntry: Rollback of journal header success.");
else
mpLogger->log(LOG_CRIT, "Replicator::addJournalEntry: Rollback of journal header failed!");
errno = l_errno;
if (err < 0)
return err;
else
return 0;
}
const size_t headerLenght = header.length();
std::memcpy(&dataStr[0], header.c_str(), headerLenght);
dataStr[headerLenght] = 0;
dataStrOffset = headerLenght + 1;
journalOffset = headerLenght + 1;
repHeaderDataWritten += headerLenght + 1;
}
std::memcpy(&dataStr[dataStrOffset], &journalData[journalOffset], journalData.size() - journalOffset);
dataStrOffset = journalData.size();
}
off_t entryHeaderOffset = ::lseek(fd, 0, SEEK_END);
err = _write(fd, offlen, JOURNAL_ENTRY_HEADER_SIZE);
l_errno = errno;
std::memcpy(&dataStr[dataStrOffset], offlen, JOURNAL_ENTRY_HEADER_SIZE);
dataStrOffset += JOURNAL_ENTRY_HEADER_SIZE;
repHeaderDataWritten += JOURNAL_ENTRY_HEADER_SIZE;
if (err != JOURNAL_ENTRY_HEADER_SIZE)
std::memcpy(&dataStr[dataStrOffset], data, length);
dataStrOffset += length;
assert(dataStr.size() == dataStrOffset);
if (journalExists && !journalHandler.removeBlob(kvStorage, journalName))
{
// this entry failed so if the header was updated roll it back
if (bHeaderChanged)
{
mpLogger->log(LOG_CRIT,
"Replicator::addJournalEntry: write journal entry header failed. Attempting to rollback "
"and continue.");
// attempt to rollback top level header
int rollbackErr = _pwrite(fd, headerRollback.c_str(), headerRollback.length() + 1, 0);
if ((uint)rollbackErr != (headerRollback.length() + 1))
{
mpLogger->log(LOG_CRIT, "Replicator::addJournalEntry: Rollback of journal header failed! (%s)",
strerror_r(errno, errbuf, 80));
errno = l_errno;
if (err < 0)
return err;
else
return 0;
}
}
int rollbackErr = ::ftruncate(fd, entryHeaderOffset);
if (rollbackErr != 0)
{
mpLogger->log(LOG_CRIT, "Replicator::addJournalEntry: Truncate to previous EOF failed! (%s)",
strerror_r(errno, errbuf, 80));
errno = l_errno;
if (err < 0)
return err;
else
return 0;
}
l_errno = errno;
return err;
}
while (count < length)
{
err = _write(fd, &data[count], length - count);
if (err < 0)
{
l_errno = errno;
/* XXXBEN: Attempt to update entry header with the partial write and write it.
IF the write fails to update entry header report an error to IOC and logging.
*/
if (count > 0) // return what was successfully written
{
mpLogger->log(LOG_CRIT,
"Replicator::addJournalEntry: Got '%s' writing a journal entry. "
"Attempting to update and continue.",
strerror_r(l_errno, errbuf, 80));
// Update the file header max_offset if necessary and possible
thisEntryMaxOffset = (offset + count - 1);
if (thisEntryMaxOffset > currentMaxOffset)
{
string header = (boost::format("{ \"version\" : \"%03i\", \"max_offset\" : \"%011u\" }") % version %
thisEntryMaxOffset)
.str();
int rollbackErr = _pwrite(fd, header.c_str(), header.length() + 1, 0);
if ((uint)rollbackErr != (header.length() + 1))
{
mpLogger->log(LOG_CRIT, "Replicator::addJournalEntry: Update of journal header failed! (%s)",
strerror_r(errno, errbuf, 80));
errno = l_errno;
return err;
}
}
// Update the journal entry header
offlen[1] = count;
int rollbackErr = _pwrite(fd, offlen, JOURNAL_ENTRY_HEADER_SIZE, entryHeaderOffset);
if ((uint)rollbackErr != JOURNAL_ENTRY_HEADER_SIZE)
{
mpLogger->log(LOG_CRIT, "Replicator::addJournalEntry: Update of journal entry header failed! (%s)",
strerror_r(errno, errbuf, 80));
errno = l_errno;
return err;
}
// return back what we did write
mpLogger->log(LOG_CRIT, "Replicator::addJournalEntry: Partial write success.");
repUserDataWritten += count;
return count;
}
else
{
// If the header was changed rollback and reset EOF
// Like this never happened
// Currently since this returns the err from the first write. IOC returns -1 and writeTask returns an
// error So system is likely broken in some way
if (bHeaderChanged)
{
mpLogger->log(LOG_CRIT,
"Replicator::addJournalEntry: write journal entry failed (%s). "
"Attempting to rollback and continue.",
strerror_r(l_errno, errbuf, 80));
// attempt to rollback top level header
string header =
(boost::format("{ \"version\" : \"%03i\", \"max_offset\" : \"%011u\" }") % version % 0).str();
int rollbackErr = _pwrite(fd, header.c_str(), header.length() + 1, 0);
if ((uint)rollbackErr != (header.length() + 1))
{
mpLogger->log(LOG_CRIT, "Replicator::addJournalEntry: Rollback of journal header failed (%s)!",
strerror_r(errno, errbuf, 80));
errno = l_errno;
return err;
}
}
int rollbackErr = ::ftruncate(fd, entryHeaderOffset);
if (rollbackErr != 0)
{
mpLogger->log(LOG_CRIT, "Replicator::addJournalEntry: Remove entry header failed (%s)!",
strerror_r(errno, errbuf, 80));
errno = l_errno;
return err;
}
mpLogger->log(LOG_CRIT, "Replicator::addJournalEntry: Write failed. Journal file restored.");
errno = l_errno;
return err;
}
}
count += err;
mpLogger->log(LOG_CRIT, "Cannot remove journal blob.");
errno = EIO;
return -1;
}
repUserDataWritten += count;
return count;
if (!journalHandler.writeBlob(kvStorage, journalName, dataStr))
{
mpLogger->log(LOG_CRIT, "Cannot write journal blob.");
errno = EIO;
return -1;
}
{
auto tnx = kvStorage->createTransaction();
tnx->set(journalSizeName, std::to_string(dataStr.size()));
if (!tnx->commit())
{
mpLogger->log(LOG_CRIT, "Cannot write journal size.");
errno = EIO;
return -1;
}
}
repUserDataWritten += length;
return length;
}
int Replicator::remove(const boost::filesystem::path& filename, Flags flags)
@@ -484,6 +376,27 @@ int Replicator::remove(const boost::filesystem::path& filename, Flags flags)
return ret;
}
int Replicator::removeJournal(const boost::filesystem::path& filename)
{
auto kvStorage = KVStorageInitializer::getStorageInstance();
auto keyGen = std::make_shared<FDBCS::BoostUIDKeyGenerator>();
FDBCS::BlobHandler journalHandler(keyGen);
if (!journalHandler.removeBlob(kvStorage, filename.string()))
{
return -1;
}
return 0;
}
int Replicator::removeJournalSize(const boost::filesystem::path& filename)
{
auto kvStorage = KVStorageInitializer::getStorageInstance();
auto tnx = kvStorage->createTransaction();
tnx->remove(filename.string());
return tnx->commit();
}
int Replicator::updateMetadata(MetadataFile& meta)
{
return meta.writeMetadata();

3
storage-manager/src/Replicator.h
View File

@@ -43,13 +43,14 @@ class Replicator
LOCAL_ONLY = 0x1,
NO_LOCAL = 0x2
};
int addJournalEntry(const boost::filesystem::path& filename, const uint8_t* data, off_t offset,
size_t length);
int newObject(const boost::filesystem::path& filename, const uint8_t* data, off_t offset, size_t length);
int newNullObject(const boost::filesystem::path& filename, size_t length);
int remove(const boost::filesystem::path& file, Flags flags = NONE);
int removeJournal(const boost::filesystem::path &file);
int removeJournalSize(const boost::filesystem::path &file);
int updateMetadata(MetadataFile& meta);

114
storage-manager/src/Synchronizer.cpp
View File

@@ -20,6 +20,8 @@
#include "IOCoordinator.h"
#include "MetadataFile.h"
#include "Utilities.h"
#include "KVStorageInitializer.h"
#include "KVPrefixes.h"
#include <boost/thread/mutex.hpp>
#include <sys/stat.h>
@@ -238,8 +240,12 @@ void Synchronizer::flushObject(const bf::path& prefix, const string& _key)
sleep(5);
}
} while (err);
journalExists = bf::exists(journalPath / (key + ".journal"));
const auto jounralName = getJournalName((journalPath / (key + ".journal")).string());
auto kvStorage = KVStorageInitializer::getStorageInstance();
auto tnx = kvStorage->createTransaction();
auto resultPair = tnx->get(jounralName);
journalExists = resultPair.first;
if (journalExists)
{
logger->log(LOG_DEBUG,
@@ -292,8 +298,8 @@ void Synchronizer::periodicSync()
else
++flushesTriggeredByTimer;
}
// cout << "Sync'ing " << pendingOps.size() << " objects" << " queue size is " <<
// threadPool.currentQueueSize() << endl;
// cout << "Sync'ing " << pendingOps.size() << " objects"
// << " queue size is " << threadPool->currentQueueSize() << endl;
for (auto& job : pendingOps)
makeJob(job.first);
for (auto it = uncommittedJournalSize.begin(); it != uncommittedJournalSize.end(); ++it)
@@ -575,11 +581,18 @@ void Synchronizer::synchronizeWithJournal(const string& sourceFile, list<string>
cache->deletedObject(prefix, cloudKey, objSize);
cs->deleteObject(cloudKey);
}
bf::path jPath = journalPath / (key + ".journal");
if (bf::exists(jPath))
const auto journalName = getJournalName((journalPath / (key + ".journal")).string());
const auto journalSizeName = getJournalName((journalPath / (key + "_size.journal")).string());
auto kvStorage = KVStorageInitializer::getStorageInstance();
auto tnx = kvStorage->createTransaction();
auto resultPairJournal = tnx->get(journalName);
auto resultPairJournalSize = tnx->get(journalName);
if (resultPairJournal.first && resultPairJournalSize.first)
{
size_t jSize = bf::file_size(jPath);
replicator->remove(jPath);
size_t jSize = std::atoi(resultPairJournalSize.second.c_str());
replicator->removeJournal(journalName);
replicator->removeJournalSize(journalSizeName);
cache->deletedJournal(prefix, jSize);
}
}
@@ -605,43 +618,48 @@ void Synchronizer::synchronizeWithJournal(const string& sourceFile, list<string>
// sync queue
bf::path oldCachePath = cachePath / key;
string journalName = (journalPath / (key + ".journal")).string();
if (!bf::exists(journalName))
const string journalName = getJournalName((journalPath / (key + ".journal")).string());
const auto journalSizeName = getJournalName((journalPath / (key + "_size.journal")).string());
{
logger->log(LOG_DEBUG, "synchronizeWithJournal(): no journal file found for %s", key.c_str());
// sanity check + add'l info. Test whether the object exists in cloud storage. If so, complain,
// and run synchronize() instead.
bool existsOnCloud;
int err = cs->exists(cloudKey, &existsOnCloud);
if (err)
throw runtime_error(string("Synchronizer: cs->exists() failed: ") + strerror_r(errno, buf, 80));
if (!existsOnCloud)
auto kvStorage = KVStorageInitializer::getStorageInstance();
auto tnx = kvStorage->createTransaction();
auto resultPair = tnx->get(journalName);
if (!resultPair.first)
{
if (cache->exists(prefix, cloudKey))
logger->log(LOG_DEBUG, "synchronizeWithJournal(): no journal file found for %s", key.c_str());
// sanity check + add'l info. Test whether the object exists in cloud storage. If so, complain,
// and run synchronize() instead.
bool existsOnCloud;
int err = cs->exists(cloudKey, &existsOnCloud);
if (err)
throw runtime_error(string("Synchronizer: cs->exists() failed: ") + strerror_r(errno, buf, 80));
if (!existsOnCloud)
{
logger->log(LOG_DEBUG,
"synchronizeWithJournal(): %s has no journal and does not exist in the cloud, calling "
"synchronize() instead. Need to explain how this happens.",
key.c_str());
s.unlock();
synchronize(sourceFile, lit);
if (cache->exists(prefix, cloudKey))
{
logger->log(LOG_DEBUG,
"synchronizeWithJournal(): %s has no journal and does not exist in the cloud, calling "
"synchronize() instead. Need to explain how this happens.",
key.c_str());
s.unlock();
synchronize(sourceFile, lit);
}
else
logger->log(LOG_DEBUG,
"synchronizeWithJournal(): %s has no journal, and does not exist in the cloud or in "
" the local cache. Need to explain how this happens.",
key.c_str());
return;
}
else
logger->log(LOG_DEBUG,
"synchronizeWithJournal(): %s has no journal, and does not exist in the cloud or in "
" the local cache. Need to explain how this happens.",
"synchronizeWithJournal(): %s has no journal, but it does exist in the cloud. "
" This indicates that an overlapping syncWithJournal() call handled it properly.",
key.c_str());
return;
}
else
logger->log(LOG_DEBUG,
"synchronizeWithJournal(): %s has no journal, but it does exist in the cloud. "
" This indicates that an overlapping syncWithJournal() call handled it properly.",
key.c_str());
return;
}
int err;
@@ -649,7 +667,6 @@ void Synchronizer::synchronizeWithJournal(const string& sourceFile, list<string>
size_t count = 0, size = mdEntry.length, originalSize = 0;
bool oldObjIsCached = cache->exists(prefix, cloudKey);
// get the base object if it is not already cached
// merge it with its journal file
if (!oldObjIsCached)
@@ -685,7 +702,10 @@ void Synchronizer::synchronizeWithJournal(const string& sourceFile, list<string>
err = ioc->mergeJournalInMem(data, size, journalName.c_str(), &_bytesRead);
if (err)
{
if (!bf::exists(journalName))
auto kvStorage = KVStorageInitializer::getStorageInstance();
auto tnx = kvStorage->createTransaction();
auto resultPair = tnx->get(journalName);
if (!resultPair.first)
logger->log(LOG_DEBUG,
"synchronizeWithJournal(): journal %s was deleted mid-operation, check locking",
journalName.c_str());
@@ -704,7 +724,10 @@ void Synchronizer::synchronizeWithJournal(const string& sourceFile, list<string>
data = ioc->mergeJournal(oldCachePath.string().c_str(), journalName.c_str(), 0, size, &_bytesRead);
if (!data)
{
if (!bf::exists(journalName))
auto kvStorage = KVStorageInitializer::getStorageInstance();
auto tnx = kvStorage->createTransaction();
auto resultPair = tnx->get(journalName);
if (!resultPair.first)
logger->log(LOG_DEBUG,
"synchronizeWithJournal(): journal %s was deleted mid-operation, check locking",
journalName.c_str());
@@ -790,8 +813,17 @@ void Synchronizer::synchronizeWithJournal(const string& sourceFile, list<string>
rename(key, newKey);
// delete the old object & journal file
cache->deletedJournal(prefix, bf::file_size(journalName));
replicator->remove(journalName);
auto kvStorage = KVStorageInitializer::getStorageInstance();
{
auto tnx = kvStorage->createTransaction();
auto result = tnx->get(journalSizeName);
if (result.first)
cache->deletedJournal(prefix, std::atoi(result.second.c_str()));
}
// Remove journal size name from kv storage.
replicator->removeJournalSize(journalSizeName);
replicator->removeJournal(journalName);
cs->deleteObject(cloudKey);
}
@@ -887,7 +919,7 @@ void Synchronizer::configListener()
{
maxUploads = 20;
threadPool->setMaxThreads(maxUploads);
logger->log(LOG_INFO, "max_concurrent_uploads = %u",maxUploads);
logger->log(LOG_INFO, "max_concurrent_uploads = %u", maxUploads);
}
if (stmp.empty())
{