database/mariadb-columnstore-engine
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2025-07-27 21:01:50 +03:00
Code Activity

Incremental commit. Stripping out the OBE fstream IO code.

Browse Source
This commit is contained in:
Patrick LeBlanc
2019-07-12 10:06:56 -05:00
parent 3e77912dcd
commit 16dc887f20
8 changed files with 256 additions and 922 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
versioning/BRM/blockresolutionmanager.cpp
View File

@ -108,21 +108,9 @@ int BlockResolutionManager::saveState(string filename) throw()
// truncate teh file if already exists since no truncate in HDFS. // truncate teh file if already exists since no truncate in HDFS.
const char* filename = journalFilename.c_str(); const char* filename = journalFilename.c_str();
IDBDataFile* journal = IDBDataFile::open(
if (true || IDBPolicy::useHdfs()) IDBPolicy::getType(filename, IDBPolicy::WRITEENG), filename, "wb", 0);
{ delete journal;
IDBDataFile* journal = IDBDataFile::open(
IDBPolicy::getType(filename, IDBPolicy::WRITEENG), filename, "wb", 0);
delete journal;
}
else
{
ofstream journal;
uint32_t utmp = ::umask(0);
journal.open(filename, ios_base::out | ios_base::trunc | ios_base::binary);
journal.close();
::umask(utmp);
}
vbbm.save(vbbmFilename); vbbm.save(vbbmFilename);
vss.save(vssFilename); vss.save(vssFilename);

380
versioning/BRM/extentmap.cpp
View File

@ -1133,85 +1133,6 @@ void ExtentMap::reserveLBIDRange(LBID_t start, uint8_t size)
... (* numFL) ... (* numFL)
*/ */
void ExtentMap::loadVersion4(ifstream& in)
{
int emNumElements, flNumElements;
in.read((char*) &emNumElements, sizeof(int));
in.read((char*) &flNumElements, sizeof(int));
idbassert(emNumElements > 0);
void *fExtentMapPtr = static_cast<void*>(fExtentMap);
memset(fExtentMapPtr, 0, fEMShminfo->allocdSize);
fEMShminfo->currentSize = 0;
// init the free list
memset(fFreeList, 0, fFLShminfo->allocdSize);
fFreeList[0].size = (1 << 26); // 2^36 LBIDs
fFLShminfo->currentSize = sizeof(InlineLBIDRange);
// @Bug 3498
// Calculate how big an extent map we're going to need and allocate it in one call
if ((fEMShminfo->allocdSize / sizeof(EMEntry)) < (unsigned)emNumElements)
{
size_t nrows = emNumElements;
//Round up to the nearest EM_INCREMENT_ROWS
if ((nrows % EM_INCREMENT_ROWS) != 0)
{
nrows /= EM_INCREMENT_ROWS;
nrows++;
nrows *= EM_INCREMENT_ROWS;
}
growEMShmseg(nrows);
}
for (int i = 0; i < emNumElements; i++)
{
in.read((char*) &fExtentMap[i], sizeof(EMEntry));
reserveLBIDRange(fExtentMap[i].range.start, fExtentMap[i].range.size);
//@bug 1911 - verify status value is valid
if (fExtentMap[i].status < EXTENTSTATUSMIN ||
fExtentMap[i].status > EXTENTSTATUSMAX)
fExtentMap[i].status = EXTENTAVAILABLE;
}
fEMShminfo->currentSize = emNumElements * sizeof(EMEntry);
#ifdef DUMP_EXTENT_MAP
EMEntry* emSrc = fExtentMap;
cout << "lbid\tsz\toid\tfbo\thwm\tpart#\tseg#\tDBRoot\twid\tst\thi\tlo\tsq\tv" << endl;
for (int i = 0; i < emNumElements; i++)
{
cout <<
emSrc[i].start
<< '\t' << emSrc[i].size
<< '\t' << emSrc[i].fileID
<< '\t' << emSrc[i].blockOffset
<< '\t' << emSrc[i].HWM
<< '\t' << emSrc[i].partitionNum
<< '\t' << emSrc[i].segmentNum
<< '\t' << emSrc[i].dbRoot
<< '\t' << emSrc[i].status
<< '\t' << emSrc[i].partition.cprange.hi_val
<< '\t' << emSrc[i].partition.cprange.lo_val
<< '\t' << emSrc[i].partition.cprange.sequenceNum
<< '\t' << (int)(emSrc[i].partition.cprange.isValid)
<< endl;
}
cout << "Free list entries:" << endl;
cout << "start\tsize" << endl;
for (int i = 0; i < flNumElements; i++)
cout << fFreeList[i].start << '\t' << fFreeList[i].size << endl;
#endif
}
void ExtentMap::loadVersion4(IDBDataFile* in) void ExtentMap::loadVersion4(IDBDataFile* in)
{ {
int emNumElements = 0, flNumElements = 0; int emNumElements = 0, flNumElements = 0;
@ -1337,79 +1258,37 @@ void ExtentMap::load(const string& filename, bool fixFL)
throw; throw;
} }
// XXXPAT: Forcing the IDB path. Remove the fstream path once we see this works. const char* filename_p = filename.c_str();
if (true || IDBPolicy::useHdfs()) scoped_ptr<IDBDataFile> in(IDBDataFile::open(
IDBPolicy::getType(filename_p, IDBPolicy::WRITEENG),
filename_p, "r", 0));
if (!in)
{ {
const char* filename_p = filename.c_str(); log_errno("ExtentMap::load(): open");
scoped_ptr<IDBDataFile> in(IDBDataFile::open( releaseFreeList(WRITE);
IDBPolicy::getType(filename_p, IDBPolicy::WRITEENG), releaseEMEntryTable(WRITE);
filename_p, "r", 0)); throw ios_base::failure("ExtentMap::load(): open failed. Check the error log.");
}
if (!in) try
{ {
log_errno("ExtentMap::load(): open"); int emVersion = 0;
releaseFreeList(WRITE); int bytes = in->read((char*) &emVersion, sizeof(int));
releaseEMEntryTable(WRITE);
throw ios_base::failure("ExtentMap::load(): open failed. Check the error log.");
}
try if (bytes == (int) sizeof(int) && emVersion == EM_MAGIC_V4)
loadVersion4(in.get());
else
{ {
int emVersion = 0; log("ExtentMap::load(): That file is not a valid ExtentMap image");
int bytes = in->read((char*) &emVersion, sizeof(int)); throw runtime_error("ExtentMap::load(): That file is not a valid ExtentMap image");
if (bytes == (int) sizeof(int) && emVersion == EM_MAGIC_V4)
loadVersion4(in.get());
else
{
log("ExtentMap::load(): That file is not a valid ExtentMap image");
throw runtime_error("ExtentMap::load(): That file is not a valid ExtentMap image");
}
}
catch (...)
{
releaseFreeList(WRITE);
releaseEMEntryTable(WRITE);
throw;
} }
} }
else // fstream path to be remove catch (...)
{ {
ifstream in; releaseFreeList(WRITE);
in.open(filename.c_str(), ios_base::in | ios_base::binary); releaseEMEntryTable(WRITE);
throw;
if (!in)
{
log_errno("ExtentMap::load(): open");
releaseFreeList(WRITE);
releaseEMEntryTable(WRITE);
throw ios_base::failure("ExtentMap::load(): open failed. Check the error log.");
}
in.exceptions(ios_base::badbit | ios_base::failbit);
try
{
int emVersion;
in.read((char*) &emVersion, sizeof(int));
if (emVersion == EM_MAGIC_V4)
loadVersion4(in);
else
{
log("ExtentMap::load(): That file is not a valid ExtentMap image");
throw runtime_error("ExtentMap::load(): That file is not a valid ExtentMap image");
}
}
catch (...)
{
in.close();
releaseFreeList(WRITE);
releaseEMEntryTable(WRITE);
throw;
}
in.close();
} }
releaseFreeList(WRITE); releaseFreeList(WRITE);
@ -1453,77 +1332,53 @@ void ExtentMap::save(const string& filename)
throw runtime_error("ExtentMap::save(): got request to save an empty BRM"); throw runtime_error("ExtentMap::save(): got request to save an empty BRM");
} }
// XXXPAT: I don't know why there are two options here. It can just use the IDBDataFile stuff. const char* filename_p = filename.c_str();
// Forcing the IDB option to execute for now. Leaving the old fstream version there in case we find there's scoped_ptr<IDBDataFile> out(IDBDataFile::open(
// a case the IDB option doesn't work. IDBPolicy::getType(filename_p, IDBPolicy::WRITEENG),
if (true || IDBPolicy::useHdfs()) filename_p, "wb", IDBDataFile::USE_VBUF));
if (!out)
{ {
const char* filename_p = filename.c_str(); log_errno("ExtentMap::save(): open");
scoped_ptr<IDBDataFile> out(IDBDataFile::open( releaseFreeList(READ);
IDBPolicy::getType(filename_p, IDBPolicy::WRITEENG), releaseEMEntryTable(READ);
filename_p, "wb", IDBDataFile::USE_VBUF)); throw ios_base::failure("ExtentMap::save(): open failed. Check the error log.");
}
if (!out) loadSize[0] = EM_MAGIC_V4;
loadSize[1] = fEMShminfo->currentSize / sizeof(EMEntry);
loadSize[2] = fFLShminfo->allocdSize / sizeof(InlineLBIDRange); // needs to send all entries
int bytes = 0;
try
{
const int wsize = 3 * sizeof(int);
bytes = out->write((char*)loadSize, wsize);
if (bytes != wsize)
throw ios_base::failure("ExtentMap::save(): write failed. Check the error log.");
}
catch (...)
{
releaseFreeList(READ);
releaseEMEntryTable(READ);
throw;
}
allocdSize = fEMShminfo->allocdSize / sizeof(EMEntry);
//const int emEntrySize = sizeof(EMEntry);
int first = -1, last = -1, err;
size_t progress, writeSize;
for (i = 0; i < allocdSize; i++)
{
if (fExtentMap[i].range.size > 0 && first == -1)
first = i;
else if (fExtentMap[i].range.size <= 0 && first != -1)
{ {
log_errno("ExtentMap::save(): open"); last = i;
releaseFreeList(READ); writeSize = (last - first) * sizeof(EMEntry);
releaseEMEntryTable(READ);
throw ios_base::failure("ExtentMap::save(): open failed. Check the error log.");
}
loadSize[0] = EM_MAGIC_V4;
loadSize[1] = fEMShminfo->currentSize / sizeof(EMEntry);
loadSize[2] = fFLShminfo->allocdSize / sizeof(InlineLBIDRange); // needs to send all entries
int bytes = 0;
try
{
const int wsize = 3 * sizeof(int);
bytes = out->write((char*)loadSize, wsize);
if (bytes != wsize)
throw ios_base::failure("ExtentMap::save(): write failed. Check the error log.");
}
catch (...)
{
releaseFreeList(READ);
releaseEMEntryTable(READ);
throw;
}
allocdSize = fEMShminfo->allocdSize / sizeof(EMEntry);
//const int emEntrySize = sizeof(EMEntry);
int first = -1, last = -1, err;
size_t progress, writeSize;
for (i = 0; i < allocdSize; i++)
{
if (fExtentMap[i].range.size > 0 && first == -1)
first = i;
else if (fExtentMap[i].range.size <= 0 && first != -1)
{
last = i;
writeSize = (last - first) * sizeof(EMEntry);
progress = 0;
char *writePos = (char *) &fExtentMap[first];
while (progress < writeSize)
{
err = out->write(writePos + progress, writeSize - progress);
if (err < 0)
{
releaseFreeList(READ);
releaseEMEntryTable(READ);
throw ios_base::failure("ExtentMap::save(): write failed. Check the error log.");
}
progress += err;
}
first = -1;
}
}
if (first != -1)
{
writeSize = (allocdSize - first) * sizeof(EMEntry);
progress = 0; progress = 0;
char *writePos = (char *) &fExtentMap[first]; char *writePos = (char *) &fExtentMap[first];
while (progress < writeSize) while (progress < writeSize)
@ -1537,14 +1392,14 @@ void ExtentMap::save(const string& filename)
} }
progress += err; progress += err;
} }
first = -1;
} }
}
//allocdSize = fFLShminfo->allocdSize / sizeof(InlineLBIDRange); if (first != -1)
//const int inlineLbidRangeSize = sizeof(InlineLBIDRange); {
writeSize = (allocdSize - first) * sizeof(EMEntry);
progress = 0; progress = 0;
writeSize = fFLShminfo->allocdSize; char *writePos = (char *) &fExtentMap[first];
char *writePos = (char *) fFreeList;
while (progress < writeSize) while (progress < writeSize)
{ {
err = out->write(writePos + progress, writeSize - progress); err = out->write(writePos + progress, writeSize - progress);
@ -1557,86 +1412,23 @@ void ExtentMap::save(const string& filename)
progress += err; progress += err;
} }
} }
else // this is the fstream version to be expired
//allocdSize = fFLShminfo->allocdSize / sizeof(InlineLBIDRange);
//const int inlineLbidRangeSize = sizeof(InlineLBIDRange);
progress = 0;
writeSize = fFLShminfo->allocdSize;
char *writePos = (char *) fFreeList;
while (progress < writeSize)
{ {
ofstream out; err = out->write(writePos + progress, writeSize - progress);
if (err < 0)
// Make em writes to disk use a buffer size of StrmBufSize bytes (instead of the default 8K)
const unsigned StrmBufSize = 1 * 1024 * 1024;
scoped_array<char> buf(new char[StrmBufSize]);
out.rdbuf()->pubsetbuf(buf.get(), StrmBufSize);
utmp = ::umask(0);
out.open(filename.c_str(), ios_base::out | ios_base::binary);
::umask(utmp);
if (!out)
{ {
log_errno("ExtentMap::save(): open");
releaseFreeList(READ); releaseFreeList(READ);
releaseEMEntryTable(READ); releaseEMEntryTable(READ);
throw ios_base::failure("ExtentMap::save(): open failed. Check the error log."); throw ios_base::failure("ExtentMap::save(): write failed. Check the error log.");
} }
progress += err;
out.exceptions(ios_base::badbit);
loadSize[0] = EM_MAGIC_V4;
loadSize[1] = fEMShminfo->currentSize / sizeof(EMEntry);
loadSize[2] = fFLShminfo->allocdSize / sizeof(InlineLBIDRange); // needs to send all entries
try
{
out.write((char*)loadSize, 3 * sizeof(int));
}
catch (...)
{
out.close();
releaseFreeList(READ);
releaseEMEntryTable(READ);
throw;
}
allocdSize = fEMShminfo->allocdSize / sizeof(EMEntry);
for (i = 0; i < allocdSize; i++)
{
if (fExtentMap[i].range.size > 0)
{
try
{
out.write((char*) &fExtentMap[i], sizeof(EMEntry));
}
catch (...)
{
out.close();
releaseFreeList(READ);
releaseEMEntryTable(READ);
throw;
}
}
}
allocdSize = fFLShminfo->allocdSize / sizeof(InlineLBIDRange);
for (i = 0; i < allocdSize; i++)
{
// if (fFreeList[i].size > 0) {
try
{
out.write((char*) &fFreeList[i], sizeof(InlineLBIDRange));
}
catch (...)
{
out.close();
releaseFreeList(READ);
releaseEMEntryTable(READ);
throw;
}
// }
}
out.close();
} }
releaseFreeList(READ); releaseFreeList(READ);

343
versioning/BRM/oidserver.cpp
View File

@ -117,34 +117,6 @@ namespace BRM
boost::mutex OIDServer::fMutex; boost::mutex OIDServer::fMutex;
#if 0
void OIDServer::lockFile() const
{
int err, errCount;
int errnoSave = 0;
for (errCount = 0, err = -1; err != 0 && errCount < MaxRetries;)
{
err = flock(fFd, LOCK_EX);
if (err < 0 && errno != EINTR) // EINTR isn't really an error
{
errCount++;
errnoSave = errno; // save errno because perror may overwrite
perror("OIDServer::lockFile(): flock (retrying)");
}
}
if (errCount == MaxRetries)
{
ostringstream oss;
oss << "OIDServer::lockFile(): flock error: " << strerror(errnoSave);
throw ios_base::failure(oss.str());
}
}
#endif
void OIDServer::writeData(uint8_t* buf, off_t offset, int size) const void OIDServer::writeData(uint8_t* buf, off_t offset, int size) const
{ {
int errCount, err, progress; int errCount, err, progress;
@ -153,73 +125,39 @@ void OIDServer::writeData(uint8_t* buf, off_t offset, int size) const
if (size == 0) if (size == 0)
return; return;
// XXXPAT: Forcing the IDB* path. Get rid of the fstream path when appropriate. for (errCount = 0; errCount < MaxRetries && seekerr != offset; errCount++)
if (true || IDBPolicy::useHdfs())
{ {
for (errCount = 0; errCount < MaxRetries && seekerr != offset; errCount++) seekerr = fFp->seek(offset, SEEK_SET);
{
seekerr = fFp->seek(offset, SEEK_SET);
if (seekerr >= 0) if (seekerr >= 0)
seekerr = fFp->tell(); // IDBDataFile may use fseek for seek. seekerr = fFp->tell(); // IDBDataFile may use fseek for seek.
if (seekerr < 0) if (seekerr < 0)
perror("OIDServer::writeDataHdfs(): lseek"); perror("OIDServer::writeData(): lseek");
}
if (errCount == MaxRetries)
throw ios_base::failure("OIDServer::writeDataHdfs(): lseek failed "
"too many times");
for (progress = 0, errCount = 0; progress < size && errCount < MaxRetries;)
{
err = fFp->write(&buf[progress], size - progress);
if (err < 0)
{
if (errno != EINTR) // EINTR isn't really an error
{
errCount++;
perror("OIDServer::writeDataHdfs(): write (retrying)");
}
}
else
progress += err;
}
fFp->tell();
} }
else
if (errCount == MaxRetries)
throw ios_base::failure("OIDServer::writeData(): lseek failed "
"too many times");
for (progress = 0, errCount = 0; progress < size && errCount < MaxRetries;)
{ {
for (errCount = 0; errCount < MaxRetries && seekerr != offset; errCount++) err = fFp->write(&buf[progress], size - progress);
if (err < 0)
{ {
seekerr = lseek(fFd, offset, SEEK_SET); if (errno != EINTR) // EINTR isn't really an error
if (seekerr < 0)
perror("OIDServer::writeData(): lseek");
}
if (errCount == MaxRetries)
throw ios_base::failure("OIDServer::writeData(): lseek failed "
"too many times");
for (progress = 0, errCount = 0; progress < size && errCount < MaxRetries;)
{
err = write(fFd, &buf[progress], size - progress);
if (err < 0)
{ {
if (errno != EINTR) // EINTR isn't really an error errCount++;
{ perror("OIDServer::writeData(): write (retrying)");
errCount++;
perror("OIDServer::writeData(): write (retrying)");
}
} }
else
progress += err;
} }
else
progress += err;
} }
fFp->tell();
if (errCount == MaxRetries) if (errCount == MaxRetries)
throw ios_base::failure("OIDServer::writeData(): write error"); throw ios_base::failure("OIDServer::writeData(): write error");
} }
@ -232,73 +170,37 @@ void OIDServer::readData(uint8_t* buf, off_t offset, int size) const
if (size == 0) if (size == 0)
return; return;
// XXXPAT: Forcing the IDB* path. Get rid of the fstream path when appropriate. for (errCount = 0; errCount < MaxRetries && seekerr != offset; errCount++)
if (true || IDBPolicy::useHdfs())
{ {
for (errCount = 0; errCount < MaxRetries && seekerr != offset; errCount++) seekerr = fFp->seek(offset, SEEK_SET);
{
seekerr = fFp->seek(offset, SEEK_SET);
if (seekerr >= 0) if (seekerr >= 0)
seekerr = fFp->tell(); // IDBDataFile may use fseek for seek. seekerr = fFp->tell(); // IDBDataFile may use fseek for seek.
if (seekerr < 0) if (seekerr < 0)
perror("OIDServer::readDataHdfs(): lseek"); perror("OIDServer::readData(): lseek");
}
if (errCount == MaxRetries)
throw ios_base::failure("OIDServer::readDataHdfs(): lseek failed "
"too many times");
for (progress = 0, errCount = 0; progress < size && errCount < MaxRetries;)
{
err = fFp->read(&buf[progress], size - progress);
if (err < 0)
{
if (errno != EINTR) // EINTR isn't really an error
{
errCount++;
perror("OIDServer::readDataHdfs(): read (retrying)");
}
}
else if (err == 0)
throw EOFException();
else
progress += err;
}
} }
else
if (errCount == MaxRetries)
throw ios_base::failure("OIDServer::readData(): lseek failed "
"too many times");
for (progress = 0, errCount = 0; progress < size && errCount < MaxRetries;)
{ {
for (errCount = 0; errCount < MaxRetries && seekerr != offset; errCount++) err = fFp->read(&buf[progress], size - progress);
if (err < 0)
{ {
seekerr = lseek(fFd, offset, SEEK_SET); if (errno != EINTR) // EINTR isn't really an error
if (seekerr < 0)
perror("OIDServer::readData(): lseek");
}
if (errCount == MaxRetries)
throw ios_base::failure("OIDServer::readData(): lseek failed "
"too many times");
for (progress = 0, errCount = 0; progress < size && errCount < MaxRetries;)
{
err = read(fFd, &buf[progress], size - progress);
if (err < 0)
{ {
if (errno != EINTR) // EINTR isn't really an error errCount++;
{ perror("OIDServer::readData(): read (retrying)");
errCount++;
perror("OIDServer::readData(): read (retrying)");
}
} }
else if (err == 0)
throw EOFException();
else
progress += err;
} }
else if (err == 0)
throw EOFException();
else
progress += err;
} }
if (errCount == MaxRetries) if (errCount == MaxRetries)
@ -349,14 +251,8 @@ void OIDServer::initializeBitmap() const
} }
writeData(buf, 0, HeaderSize); writeData(buf, 0, HeaderSize);
// reset buf to all 0's and write the bitmap
//for (i = 0; i < HeaderSize; i++)
// buf[i] = 0;
//for (i = 0; i < bitmapSize; i += HeaderSize)
// writeData(buf, HeaderSize + i, (bitmapSize - i > HeaderSize ? HeaderSize : bitmapSize - i));
// write the new bitmap file
uint8_t *bitmapbuf = new uint8_t[bitmapSize]; uint8_t *bitmapbuf = new uint8_t[bitmapSize];
memset(bitmapbuf, 0, bitmapSize); memset(bitmapbuf, 0, bitmapSize);
writeData(bitmapbuf, HeaderSize, bitmapSize); writeData(bitmapbuf, HeaderSize, bitmapSize);
@ -388,118 +284,55 @@ OIDServer::OIDServer() : fFp(NULL), fFd(-1)
throw runtime_error(os.str()); throw runtime_error(os.str());
} }
// XXXPAT: Forcing the IDB* path. if (!IDBPolicy::exists(fFilename.c_str())) //no bitmap file
if (true || IDBPolicy::useHdfs())
{ {
if (!IDBPolicy::exists(fFilename.c_str())) //no bitmap file BRM::DBRM em;
if (!em.isEMEmpty())
{ {
BRM::DBRM em; os << "Extent Map not empty and " << fFilename << " not found. Setting system to read-only";
cerr << os.str() << endl;
log(os.str());
em.setReadOnly(true);
throw runtime_error(os.str());
}
if (!em.isEMEmpty()) fFp = IDBDataFile::open(IDBPolicy::getType(fFilename.c_str(), IDBPolicy::WRITEENG),
{ fFilename.c_str(), "w+b", 0, 1);
os << "Extent Map not empty and " << fFilename << " not found. Setting system to read-only";
cerr << os.str() << endl;
log(os.str());
em.setReadOnly(true);
throw runtime_error(os.str());
}
fFp = IDBDataFile::open(IDBPolicy::getType(fFilename.c_str(), IDBPolicy::WRITEENG), if (!fFp)
fFilename.c_str(), "w+b", 0, 1); {
os << "Couldn't create oid bitmap file " << fFilename << ": " <<
if (!fFp) strerror(errno);
{ log(os.str());
os << "Couldn't create oid bitmap file " << fFilename << ": " << throw ios_base::failure(os.str());
strerror(errno); }
log(os.str());
throw ios_base::failure(os.str());
}
#ifndef _MSC_VER #ifndef _MSC_VER
if (fFp)
//FIXME: chmod(fFilename.c_str(), 0664); // XXXPAT: override umask at least for testing
//fchmod(fFd, 0666); // XXXPAT: override umask at least for testing
if (fFp)
chmod(fFilename.c_str(), 0664); // XXXPAT: override umask at least for testing
#endif #endif
try try
{
initializeBitmap();
}
catch (...)
{
delete fFp;
fFp = NULL;
throw;
}
}
else
{ {
fFp = IDBDataFile::open(IDBPolicy::getType(fFilename.c_str(), IDBPolicy::WRITEENG), initializeBitmap();
fFilename.c_str(), "r+b", 0, 1); }
catch (...)
if (!fFp) {
{ delete fFp;
ostringstream os; fFp = NULL;
os << "Couldn't open oid bitmap file" << fFilename << ": " << throw;
strerror(errno);
log(os.str());
throw ios_base::failure(os.str());
}
} }
} }
else else
{ {
if (access(fFilename.c_str(), F_OK) != 0) //no bitmap file fFp = IDBDataFile::open(IDBPolicy::getType(fFilename.c_str(), IDBPolicy::WRITEENG),
fFilename.c_str(), "r+b", 0, 1);
if (!fFp)
{ {
BRM::DBRM em; ostringstream os;
os << "Couldn't open oid bitmap file" << fFilename << ": " <<
if (!em.isEMEmpty())
{
os << "Extent Map not empty and " << fFilename << " not found. Setting system to read-only";
cerr << os.str() << endl;
log(os.str());
em.setReadOnly(true);
throw runtime_error(os.str());
}
}
fFd = open(fFilename.c_str(), O_CREAT | O_EXCL | O_RDWR | O_BINARY, 0664);
if (fFd >= 0)
{
#ifndef _MSC_VER
//FIXME:
fchmod(fFd, 0666); // XXXPAT: override umask at least for testing
#endif
try
{
initializeBitmap();
}
catch (...)
{
close(fFd);
throw;
}
}
else if (errno == EEXIST)
{
fFd = open(fFilename.c_str(), O_RDWR | O_BINARY);
if (fFd < 0)
{
os << "Couldn't open oid bitmap file " << fFilename << ": " <<
strerror(errno);
log(os.str());
throw ios_base::failure(os.str());
}
}
else
{
os << "Couldn't create oid bitmap file " << fFilename << ": " <<
strerror(errno); strerror(errno);
log(os.str()); log(os.str());
throw ios_base::failure(os.str()); throw ios_base::failure(os.str());
@ -621,8 +454,7 @@ retry:
{ {
writeData(buf, offset, byteSize); writeData(buf, offset, byteSize);
if (true || IDBPolicy::useHdfs()) fFp->flush();
fFp->flush();
delete [] buf; delete [] buf;
return; return;
@ -668,8 +500,7 @@ retry:
{ {
writeData(buf, offset, byteSize); writeData(buf, offset, byteSize);
if (true || IDBPolicy::useHdfs()) fFp->flush();
fFp->flush();
delete [] buf; delete [] buf;
return; return;
@ -789,9 +620,7 @@ void OIDServer::patchFreelist(struct FEntry* freelist, int start, int num) const
if (changed) if (changed)
{ {
writeData(reinterpret_cast<uint8_t*>(freelist), 0, HeaderSize); writeData(reinterpret_cast<uint8_t*>(freelist), 0, HeaderSize);
fFp->flush();
if (true || IDBPolicy::useHdfs())
fFp->flush();
} }
} }
@ -818,8 +647,7 @@ int OIDServer::allocVBOID(uint16_t dbroot)
throw; throw;
} }
if (true || IDBPolicy::useHdfs()) fFp->flush();
fFp->flush();
return ret; return ret;
} }
@ -892,8 +720,7 @@ int OIDServer::allocOIDs(int num)
writeData(reinterpret_cast<uint8_t*>(freelist), 0, HeaderSize); writeData(reinterpret_cast<uint8_t*>(freelist), 0, HeaderSize);
flipOIDBlock(bestMatchBegin, num, 0); flipOIDBlock(bestMatchBegin, num, 0);
if (true || IDBPolicy::useHdfs()) fFp->flush();
fFp->flush();
return bestMatchBegin; return bestMatchBegin;
} }

20
versioning/BRM/sessionmanagerserver.cpp
View File

@ -377,25 +377,7 @@ const TxnID SessionManagerServer::newTxnID(const SID session, bool block, bool i
if (isDDL) if (isDDL)
++_sysCatVerID; ++_sysCatVerID;
if (false && !IDBPolicy::useHdfs()) saveSMTxnIDAndState();
{
int filedata[2];
filedata[0] = _verID;
filedata[1] = _sysCatVerID;
lseek(txnidfd, 0, SEEK_SET);
int err = write(txnidfd, filedata, 8);
if (err < 0)
{
perror("SessionManagerServer::newTxnID(): write(verid)");
throw runtime_error("SessionManagerServer::newTxnID(): write(verid) failed");
}
}
else
{
saveSMTxnIDAndState();
}
return ret; return ret;
} }

200
versioning/BRM/vbbm.cpp
View File

@ -21,7 +21,6 @@
****************************************************************************/ ****************************************************************************/
#include <iostream> #include <iostream>
#include <fstream>
#include <vector> #include <vector>
#include <sstream> #include <sstream>
#include <sys/types.h> #include <sys/types.h>
@ -980,55 +979,6 @@ void VBBM::setReadOnly()
* struct VBBMEntry * numEntries * struct VBBMEntry * numEntries
*/ */
void VBBM::loadVersion1(IDBDataFile* in)
{
int vbbmEntries, i;
VBBMEntry entry;
clear();
if (in->read((char*) &vbbmEntries, 4) != 4)
{
log_errno("VBBM::load()");
throw runtime_error("VBBM::load(): Failed to read entry number");
}
for (i = 0; i < vbbmEntries; i++)
{
if (in->read((char*)&entry, sizeof(entry)) != sizeof(entry))
{
log_errno("VBBM::load()");
throw runtime_error("VBBM::load(): Failed to load entry");
}
insert(entry.lbid, entry.verID, entry.vbOID, entry.vbFBO, true);
//confirmChanges();
addVBFileIfNotExists(entry.vbOID);
}
/* This will load the saved file data from 2.2, but it is not compatible with
* 3.0+. If enabled, take out the addVBFile..() call above
*/
#if 0
int dummy, nFiles;
in.read((char*) &nFiles, 4);
cout << "got nfiles = " << nFiles << endl;
in.read((char*) &dummy, 4); // an unused var in 3.0+
while (vbbm->nFiles < nFiles)
growVBBM(true); // this allocates one file, doesn't grow the main storage
in.read((char*) files, sizeof(VBFileMetadata) * nFiles);
for (i = 0; i < nFiles; i++)
cout << "file " << i << ": oid=" << files[i].OID << " size=" << files[i].fileSize
<< " offset=" << files[i].nextOffset << endl;
#endif
}
void VBBM::loadVersion2(IDBDataFile* in) void VBBM::loadVersion2(IDBDataFile* in)
{ {
int vbbmEntries; int vbbmEntries;
@ -1091,20 +1041,6 @@ void VBBM::loadVersion2(IDBDataFile* in)
for (i = 0; i < vbbmEntries; i++) for (i = 0; i < vbbmEntries; i++)
insert(loadedEntries[i].lbid, loadedEntries[i].verID, loadedEntries[i].vbOID, insert(loadedEntries[i].lbid, loadedEntries[i].verID, loadedEntries[i].vbOID,
loadedEntries[i].vbFBO, true); loadedEntries[i].vbFBO, true);
/*
for (i = 0; i < vbbmEntries; i++)
{
if (in->read((char*)&entry, sizeof(entry)) != sizeof(entry))
{
log_errno("VBBM::load()");
throw runtime_error("VBBM::load(): Failed to load entry");
}
insert(entry.lbid, entry.verID, entry.vbOID, entry.vbFBO, true);
}
*/
} }
//#include "boost/date_time/posix_time/posix_time.hpp" //#include "boost/date_time/posix_time/posix_time.hpp"
@ -1138,10 +1074,6 @@ void VBBM::load(string filename)
switch (magic) switch (magic)
{ {
case VBBM_MAGIC_V1:
loadVersion1(in.get());
break;
case VBBM_MAGIC_V2: case VBBM_MAGIC_V2:
loadVersion2(in.get()); loadVersion2(in.get());
break; break;
@ -1161,59 +1093,38 @@ void VBBM::save(string filename)
int i; int i;
int var; int var;
// XXXPAT: forcing the IDB* path. Delete the fstream path when appropriate. const char* filename_p = filename.c_str();
if (true || IDBPolicy::useHdfs()) scoped_ptr<IDBDataFile> out(IDBDataFile::open(
IDBPolicy::getType(filename_p, IDBPolicy::WRITEENG),
filename_p, "wb", IDBDataFile::USE_VBUF));
if (!out)
{ {
const char* filename_p = filename.c_str(); log_errno("VBBM::save()");
scoped_ptr<IDBDataFile> out(IDBDataFile::open( throw runtime_error("VBBM::save(): Failed to open the file");
IDBPolicy::getType(filename_p, IDBPolicy::WRITEENG), }
filename_p, "wb", IDBDataFile::USE_VBUF));
if (!out) var = VBBM_MAGIC_V2;
int bytesWritten = 0;
int bytesToWrite = 12;
bytesWritten += out->write((char*) &var, 4);
bytesWritten += out->write((char*) &vbbm->vbCurrentSize, 4);
bytesWritten += out->write((char*) &vbbm->nFiles, 4);
bytesWritten += out->write((char*) files, sizeof(VBFileMetadata) * vbbm->nFiles);
bytesToWrite += sizeof(VBFileMetadata) * vbbm->nFiles;
int first = -1, last = -1, err;
size_t progress, writeSize;
for (i = 0; i < vbbm->vbCapacity; i++)
{
if (storage[i].lbid != -1 && first == -1)
first = i;
else if (storage[i].lbid == -1 && first != -1)
{ {
log_errno("VBBM::save()"); last = i;
throw runtime_error("VBBM::save(): Failed to open the file"); writeSize = (last - first) * sizeof(VBBMEntry);
}
var = VBBM_MAGIC_V2;
int bytesWritten = 0;
int bytesToWrite = 12;
bytesWritten += out->write((char*) &var, 4);
bytesWritten += out->write((char*) &vbbm->vbCurrentSize, 4);
bytesWritten += out->write((char*) &vbbm->nFiles, 4);
bytesWritten += out->write((char*) files, sizeof(VBFileMetadata) * vbbm->nFiles);
bytesToWrite += sizeof(VBFileMetadata) * vbbm->nFiles;
int first = -1, last = -1, err;
size_t progress, writeSize;
for (i = 0; i < vbbm->vbCapacity; i++)
{
if (storage[i].lbid != -1 && first == -1)
first = i;
else if (storage[i].lbid == -1 && first != -1)
{
last = i;
writeSize = (last - first) * sizeof(VBBMEntry);
progress = 0;
char *writePos = (char *) &storage[first];
while (progress < writeSize)
{
err = out->write(writePos + progress, writeSize - progress);
if (err < 0)
{
log_errno("VBBM::save()");
throw runtime_error("VBBM::save(): Failed to write the file");
}
progress += err;
}
first = -1;
}
}
if (first != -1)
{
writeSize = (vbbm->vbCapacity - first) * sizeof(VBBMEntry);
progress = 0; progress = 0;
char *writePos = (char *) &storage[first]; char *writePos = (char *) &storage[first];
while (progress < writeSize) while (progress < writeSize)
@ -1226,51 +1137,26 @@ void VBBM::save(string filename)
} }
progress += err; progress += err;
} }
first = -1;
} }
/*
for (i = 0; i < vbbm->vbCapacity; i++)
{
if (storage[i].lbid != -1)
{
bytesToWrite += sizeof(VBBMEntry);
bytesWritten += out->write((char*)&storage[i], sizeof(VBBMEntry));
}
}
if (bytesWritten != bytesToWrite)
{
log_errno("VBBM::save()");
throw runtime_error("VBBM::save(): Failed to write the file");
}
*/
} }
else if (first != -1)
{ {
ofstream out; writeSize = (vbbm->vbCapacity - first) * sizeof(VBBMEntry);
out.open(filename.c_str(), ios_base::trunc | ios_base::out | ios_base::binary); progress = 0;
//::umask(utmp); char *writePos = (char *) &storage[first];
while (progress < writeSize)
if (!out)
{ {
log_errno("VBBM::save()"); err = out->write(writePos + progress, writeSize - progress);
throw runtime_error("VBBM::save(): Failed to open the file"); if (err < 0)
{
log_errno("VBBM::save()");
throw runtime_error("VBBM::save(): Failed to write the file");
}
progress += err;
} }
out.exceptions(ios_base::badbit);
var = VBBM_MAGIC_V2;
out.write((char*) &var, 4);
out.write((char*) &vbbm->vbCurrentSize, 4);
out.write((char*) &vbbm->nFiles, 4);
out.write((char*) files, sizeof(VBFileMetadata) * vbbm->nFiles);
for (i = 0; i < vbbm->vbCapacity; i++)
if (storage[i].lbid != -1)
out.write((char*)&storage[i], sizeof(VBBMEntry));
} }
#if 0 #if 0
cout << "saving... nfiles=" << vbbm->nFiles << "\n"; cout << "saving... nfiles=" << vbbm->nFiles << "\n";

1
versioning/BRM/vbbm.h
View File

@ -214,7 +214,6 @@ public:
EXPORT void clear(); EXPORT void clear();
EXPORT void load(std::string filename); EXPORT void load(std::string filename);
EXPORT void loadVersion1(idbdatafile::IDBDataFile* in);
EXPORT void loadVersion2(idbdatafile::IDBDataFile* in); EXPORT void loadVersion2(idbdatafile::IDBDataFile* in);
EXPORT void save(std::string filename); EXPORT void save(std::string filename);

212
versioning/BRM/vss.cpp
View File

@ -21,7 +21,6 @@
****************************************************************************/ ****************************************************************************/
#include <iostream> #include <iostream>
#include <fstream>
#include <sstream> #include <sstream>
//#define NDEBUG //#define NDEBUG
#include <cassert> #include <cassert>
@ -1294,57 +1293,36 @@ void VSS::save(string filename)
int i; int i;
struct Header header; struct Header header;
// XXXPAT: Forcing the IDB* path to run. Delete the fstream path when appropriate. const char* filename_p = filename.c_str();
if (true || IDBPolicy::useHdfs()) scoped_ptr<IDBDataFile> out(IDBDataFile::open(
IDBPolicy::getType(filename_p, IDBPolicy::WRITEENG),
filename_p, "wb", IDBDataFile::USE_VBUF));
if (!out)
{ {
const char* filename_p = filename.c_str(); log_errno("VSS::save()");
scoped_ptr<IDBDataFile> out(IDBDataFile::open( throw runtime_error("VSS::save(): Failed to open the file");
IDBPolicy::getType(filename_p, IDBPolicy::WRITEENG), }
filename_p, "wb", IDBDataFile::USE_VBUF));
if (!out) header.magic = VSS_MAGIC_V1;
{ header.entries = vss->currentSize;
log_errno("VSS::save()");
throw runtime_error("VSS::save(): Failed to open the file");
}
header.magic = VSS_MAGIC_V1; if (out->write((char*)&header, sizeof(header)) != sizeof(header))
header.entries = vss->currentSize; {
log_errno("VSS::save()");
throw runtime_error("VSS::save(): Failed to write header to the file");
}
if (out->write((char*)&header, sizeof(header)) != sizeof(header)) int first = -1, last = -1, err;
size_t progress, writeSize;
for (i = 0; i < vss->capacity; i++)
{
if (storage[i].lbid != -1 && first == -1)
first = i;
else if (storage[i].lbid == -1 && first != -1)
{ {
log_errno("VSS::save()"); last = i;
throw runtime_error("VSS::save(): Failed to write header to the file"); writeSize = (last - first) * sizeof(VSSEntry);
}
int first = -1, last = -1, err;
size_t progress, writeSize;
for (i = 0; i < vss->capacity; i++)
{
if (storage[i].lbid != -1 && first == -1)
first = i;
else if (storage[i].lbid == -1 && first != -1)
{
last = i;
writeSize = (last - first) * sizeof(VSSEntry);
progress = 0;
char *writePos = (char *) &storage[first];
while (progress < writeSize)
{
err = out->write(writePos + progress, writeSize - progress);
if (err < 0)
{
log_errno("VSS::save()");
throw runtime_error("VSS::save(): Failed to write the file");
}
progress += err;
}
first = -1;
}
}
if (first != -1)
{
writeSize = (vss->capacity - first) * sizeof(VSSEntry);
progress = 0; progress = 0;
char *writePos = (char *) &storage[first]; char *writePos = (char *) &storage[first];
while (progress < writeSize) while (progress < writeSize)
@ -1357,56 +1335,25 @@ void VSS::save(string filename)
} }
progress += err; progress += err;
} }
first = -1;
} }
/*
for (i = 0; i < vss->capacity; i++)
{
if (storage[i].lbid != -1)
{
if (out->write((char*)&storage[i], sizeof(VSSEntry)) != sizeof(VSSEntry))
{
log_errno("VSS::save()");
throw runtime_error("VSS::save(): Failed to write vss entry to the file");
}
}
}
*/
} }
else if (first != -1)
{ {
ofstream out; writeSize = (vss->capacity - first) * sizeof(VSSEntry);
out.open(filename.c_str(), ios_base::trunc | ios_base::out | ios_base::binary); progress = 0;
//::umask(utmp); char *writePos = (char *) &storage[first];
while (progress < writeSize)
if (!out)
{ {
log_errno("VSS::save()"); err = out->write(writePos + progress, writeSize - progress);
throw runtime_error("VSS::save(): Failed to open the file"); if (err < 0)
{
log_errno("VSS::save()");
throw runtime_error("VSS::save(): Failed to write the file");
}
progress += err;
} }
out.exceptions(ios_base::badbit);
header.magic = VSS_MAGIC_V1;
header.entries = vss->currentSize;
try
{
out.write((char*)&header, sizeof(header));
for (i = 0; i < vss->capacity; i++)
if (storage[i].lbid != -1)
out.write((char*)&storage[i], sizeof(VSSEntry));
}
catch (std::exception& e)
{
out.close();
throw;
}
out.close();
} }
} }
// Ideally, we;d like to get in and out of this fcn as quickly as possible. // Ideally, we;d like to get in and out of this fcn as quickly as possible.
@ -1517,94 +1464,11 @@ void VSS::load(string filename)
for (i = 0; i < header.entries; i++) for (i = 0; i < header.entries; i++)
insert(loadedEntries[i].lbid, loadedEntries[i].verID, loadedEntries[i].vbFlag, insert(loadedEntries[i].lbid, loadedEntries[i].verID, loadedEntries[i].vbFlag,
loadedEntries[i].locked, true); loadedEntries[i].locked, true);
/*
for (i = 0; i < header.entries; i++)
{
if (in->read((char*)&entry, sizeof(entry)) != sizeof(entry))
{
log_errno("VSS::load()");
throw runtime_error("VSS::load(): Failed to read entry");
}
insert(entry.lbid, entry.verID, entry.vbFlag, entry.locked, true);
}
*/
//time2 = microsec_clock::local_time(); //time2 = microsec_clock::local_time();
//cout << "done loading " << time2 << " duration: " << time2-time1 << endl; //cout << "done loading " << time2 << " duration: " << time2-time1 << endl;
} }
#ifndef __LP64__
//This code is OBE now that the structs are padded correctly
struct VSSEntry_
{
LBID_t lbid;
VER_t verID;
bool vbFlag;
bool locked;
int next;
uint32_t pad1;
};
void VSS::load64(string filename)
{
int i;
struct Header header;
struct VSSEntry_ entry;
const char* filename_p = filename.c_str();
scoped_ptr<IDBDataFile> in(IDBDataFile::open(
IDBPolicy::getType(filename_p, IDBPolicy::WRITEENG),
filename_p, "rb", 0));
if (!in)
{
log_errno("VSS::load()");
throw runtime_error("VSS::load(): Failed to open the file");
}
if (in->read((char*)&header, sizeof(header)) != sizeof(header))
{
log_errno("VSS::load()");
throw runtime_error("VSS::load(): Failed to read header");
}
if (header.magic != VSS_MAGIC_V1)
{
log("VSS::load(): Bad magic. Not a VSS file?");
throw runtime_error("VSS::load(): Bad magic. Not a VSS file?");
}
if (header.entries < 0)
{
log("VSS::load(): Bad size. Not a VSS file?");
throw runtime_error("VSS::load(): Bad size. Not a VSS file?");
}
for (i = 0; i < vss->capacity; i++)
storage[i].lbid = -1;
for (i = 0; i < vss->numHashBuckets; i++)
hashBuckets[i] = -1;
vss->currentSize = 0;
vss->lockedEntryCount = 0;
vss->LWM = 0;
for (i = 0; i < header.entries; i++)
{
if (in->read((char*)&entry, sizeof(entry)) != sizeof(entry))
{
log_errno("VSS::load()");
throw runtime_error("VSS::load(): Failed to read entry");
}
insert(entry.lbid, entry.verID, entry.vbFlag, entry.locked, true);
}
}
#endif
#ifdef BRM_DEBUG #ifdef BRM_DEBUG
// read lock // read lock
int VSS::getShmid() const int VSS::getShmid() const

4
versioning/BRM/vss.h
View File

@ -250,10 +250,6 @@ public:
EXPORT void clear(); EXPORT void clear();
EXPORT void load(std::string filename); EXPORT void load(std::string filename);
#ifndef __LP64__
//This method is OBE now that the structs are padded correctly
EXPORT void load64(std::string filename);
#endif
EXPORT void save(std::string filename); EXPORT void save(std::string filename);
#ifdef BRM_DEBUG #ifdef BRM_DEBUG