database/mariadb-columnstore-engine
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2025-12-10 22:42:30 +03:00
Code Activity

Changed the lock scope in IOC::read() to make sure journal files

Browse Source 
can't be written to while they're being read.
This commit is contained in:
Patrick LeBlanc
2019-05-23 13:43:51 -05:00
parent e4f5d30ffa
commit 8e4f8db2a2
1 changed files with 11 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
src/IOCoordinator.cpp
View File

@@ -198,7 +198,12 @@ int IOCoordinator::read(const char *filename, uint8_t *data, off_t offset, size_
objectFDs[key] = fd; objectFDs[key] = fd;
fdMinders.push_back(SharedCloser(fd)); fdMinders.push_back(SharedCloser(fd));
} }
fileLock.unlock(); //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.
// If we do have a guarantee, and I'm just not seeing it now, add the explanation here and uncomment
// the unlock() call above. For now I will let this hold the read lock for the duration.
// copy data from each object + journal into the returned data // copy data from each object + journal into the returned data
size_t count = 0; size_t count = 0;
@@ -301,7 +306,7 @@ int IOCoordinator::_write(const char *filename, const uint8_t *data, off_t offse
if ((writeLength + objectOffset) > i->length) if ((writeLength + objectOffset) > i->length)
metadata.updateEntryLength(i->offset, (writeLength + objectOffset)); metadata.updateEntryLength(i->offset, (writeLength + objectOffset));
assert(objectOffset >= 0 && objectOffset < i->length); //assert(objectOffset >= 0 && objectOffset < i->length);
cache->newJournalEntry(writeLength+JOURNAL_ENTRY_HEADER_SIZE); cache->newJournalEntry(writeLength+JOURNAL_ENTRY_HEADER_SIZE);
@@ -338,7 +343,7 @@ int IOCoordinator::_write(const char *filename, const uint8_t *data, off_t offse
if ((writeLength + objectOffset) > newObject.length) if ((writeLength + objectOffset) > newObject.length)
metadata.updateEntryLength(newObject.offset, (writeLength + objectOffset)); metadata.updateEntryLength(newObject.offset, (writeLength + objectOffset));
assert(objectOffset >= 0 && objectOffset < newObject.length); //assert(objectOffset >= 0 && objectOffset < newObject.length);
// send to replicator // send to replicator
err = replicator->newObject(newObject.key.c_str(),&data[count],objectOffset,writeLength); err = replicator->newObject(newObject.key.c_str(),&data[count],objectOffset,writeLength);
@@ -354,7 +359,7 @@ int IOCoordinator::_write(const char *filename, const uint8_t *data, off_t offse
//log error and abort //log error and abort
} }
cache->newObject(newObject.key,writeLength); cache->newObject(newObject.key,writeLength + objectOffset);
newObjectKeys.push_back(newObject.key); newObjectKeys.push_back(newObject.key);
count += writeLength; count += writeLength;
@@ -416,7 +421,7 @@ int IOCoordinator::append(const char *filename, const uint8_t *data, size_t leng
} }
metadata.updateEntryLength(i->offset, (writeLength + i->length)); metadata.updateEntryLength(i->offset, (writeLength + i->length));
assert(i->offset >= 0 && i->offset < (writeLength + i->length)); //assert(i->offset >= 0 && i->offset < (writeLength + i->length));
cache->newJournalEntry(writeLength+JOURNAL_ENTRY_HEADER_SIZE); cache->newJournalEntry(writeLength+JOURNAL_ENTRY_HEADER_SIZE);
@@ -565,7 +570,7 @@ int IOCoordinator::truncate(const char *path, size_t newSize)
else else
{ {
meta.updateEntryLength(objects[0].offset, newSize - objects[0].offset); meta.updateEntryLength(objects[0].offset, newSize - objects[0].offset);
assert(objects[0].offset >= 0 && objects[0].offset < (newSize - objects[0].offset)); //assert(objects[0].offset >= 0 && objects[0].offset < (newSize - objects[0].offset));
} }
for (uint i = 1; i < objects.size(); i++) for (uint i = 1; i < objects.size(); i++)
meta.removeEntry(objects[i].offset); meta.removeEntry(objects[i].offset);