database/mariadb-columnstore-engine
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2025-12-17 01:02:23 +03:00
Code Activity

fixes for write and finish append. add append unit_test cases.

Browse Source
This commit is contained in:
Ben Thompson
2019-04-08 16:34:48 -05:00
parent 9b82b3b0ba
commit cf678170ca
2 changed files with 116 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

138
src/IOCoordinator.cpp
View File

@@ -247,7 +247,7 @@ int IOCoordinator::write(const char *filename, const uint8_t *data, off_t offset
uint64_t count = 0; uint64_t count = 0;
uint64_t writeLength = 0; uint64_t writeLength = 0;
uint64_t dataRemaining = length; uint64_t dataRemaining = length;
uint64_t journalOffset = 0; uint64_t objectOffset = 0;
vector<metadataObject> objects; vector<metadataObject> objects;
vector<string> newObjectKeys; vector<string> newObjectKeys;
Synchronizer *synchronizer = Synchronizer::get(); // need to init sync here to break circular dependency... Synchronizer *synchronizer = Synchronizer::get(); // need to init sync here to break circular dependency...
@@ -269,86 +269,77 @@ int IOCoordinator::write(const char *filename, const uint8_t *data, off_t offset
{ {
// first object in the list so start at offset and // first object in the list so start at offset and
// write to end of oject or all the data // write to end of oject or all the data
journalOffset = offset - i->offset; objectOffset = offset - i->offset;
writeLength = min((objectSize - journalOffset),dataRemaining); writeLength = min((objectSize - objectOffset),dataRemaining);
} }
else else
{ {
// starting at beginning of next object write the rest of data // starting at beginning of next object write the rest of data
// or until object length is reached // or until object length is reached
writeLength = min(objectSize,dataRemaining); writeLength = min(objectSize,dataRemaining);
journalOffset = 0; objectOffset = 0;
} }
cache->makeSpace(writeLength+JOURNAL_ENTRY_HEADER_SIZE); cache->makeSpace(writeLength+JOURNAL_ENTRY_HEADER_SIZE);
err = replicator->addJournalEntry(i->key.c_str(),&data[count],journalOffset,writeLength); err = replicator->addJournalEntry(i->key.c_str(),&data[count],objectOffset,writeLength);
if (err <= 0) if (err <= 0)
{ {
if ((count + journalOffset) > i->length) if ((count + objectOffset) > i->length)
metadata.updateEntryLength(i->offset, (count + journalOffset)); metadata.updateEntryLength(i->offset, (count + objectOffset));
metadata.writeMetadata(filename); metadata.writeMetadata(filename);
logger->log(LOG_ERR,"IOCoordinator::write(): object failed to complete write, %u of %u bytes written.",count,length); logger->log(LOG_ERR,"IOCoordinator::write(): object failed to complete write, %u of %u bytes written.",count,length);
return count; return count;
} }
if ((writeLength + journalOffset) > i->length) if ((writeLength + objectOffset) > i->length)
metadata.updateEntryLength(i->offset, (writeLength + journalOffset)); metadata.updateEntryLength(i->offset, (writeLength + objectOffset));
cache->newJournalEntry(writeLength+JOURNAL_ENTRY_HEADER_SIZE); cache->newJournalEntry(writeLength+JOURNAL_ENTRY_HEADER_SIZE);
synchronizer->newJournalEntry(i->key); synchronizer->newJournalEntry(i->key);
count += err; count += writeLength;
dataRemaining -= err; dataRemaining -= writeLength;
} }
} }
// there is no overlapping data, or data goes beyond end of last object // there is no overlapping data, or data goes beyond end of last object
while (dataRemaining > 0 && err >= 0) while (dataRemaining > 0)
{ {
//add a new metaDataObject cache->makeSpace(dataRemaining);
if (count == 0) metadataObject newObject = metadata.addMetadataObject(filename,0);
if (count == 0 && (uint64_t) offset > newObject.offset)
{ {
//this is starting beyond last object in metadata //this is starting beyond last object in metadata
//figure out if the offset is in this object //figure out if the offset is in this object
if ((uint64_t) offset < objectSize) objectOffset = offset - newObject.offset;
{ writeLength = min((objectSize - objectOffset),dataRemaining);
journalOffset = offset;
writeLength = min((objectSize - journalOffset),dataRemaining);
}
else
{
//we need to create an object that is only padding
}
} }
else else
{ {
// count != 0 we've already started writing and are going to new object // count != 0 we've already started writing and are going to new object
// start at beginning of the new object // start at beginning of the new object
writeLength = min(objectSize,dataRemaining); writeLength = min(objectSize,dataRemaining);
journalOffset = 0; objectOffset = 0;
} }
cache->makeSpace(journalOffset+writeLength); if ((writeLength + objectOffset) > newObject.length)
// add a new metadata object, this will get a new objectKey metadata.updateEntryLength(newObject.offset, (writeLength + objectOffset));
metadataObject newObject = metadata.addMetadataObject(filename,writeLength); // send to replicator
// write the new object err = replicator->newObject(newObject.key.c_str(),data,objectOffset,writeLength);
err = replicator->newObject(newObject.key.c_str(),data,journalOffset,writeLength);
if (err <= 0) if (err <= 0)
{ {
// update metadataObject length to reflect what awas actually written // update metadataObject length to reflect what awas actually written
if ((count + journalOffset) > newObject.length) if ((count + objectOffset) > newObject.length)
metadata.updateEntryLength(newObject.offset, (count + journalOffset)); metadata.updateEntryLength(newObject.offset, (count + objectOffset));
metadata.writeMetadata(filename); metadata.writeMetadata(filename);
logger->log(LOG_ERR,"IOCoordinator::write(): newObject failed to complete write, %u of %u bytes written.",count,length); logger->log(LOG_ERR,"IOCoordinator::write(): newObject failed to complete write, %u of %u bytes written.",count,length);
return count; return count;
//log error and abort //log error and abort
} }
if ((writeLength + journalOffset) > newObject.length)
metadata.updateEntryLength(newObject.offset, (writeLength + journalOffset));
cache->newObject(newObject.key,(writeLength + journalOffset)); cache->newObject(newObject.key,(writeLength + objectOffset));
newObjectKeys.push_back(newObject.key); newObjectKeys.push_back(newObject.key);
count += err; count += writeLength;
dataRemaining -= err; dataRemaining -= writeLength;
} }
synchronizer->newObjects(newObjectKeys); synchronizer->newObjects(newObjectKeys);
@@ -368,10 +359,11 @@ int IOCoordinator::append(const char *filename, const uint8_t *data, size_t leng
size_t count = 0; size_t count = 0;
uint64_t writeLength = 0; uint64_t writeLength = 0;
uint64_t dataRemaining = length; uint64_t dataRemaining = length;
uint64_t journalOffset = 0;
vector<metadataObject> objects; vector<metadataObject> objects;
vector<string> newObjectKeys;
Synchronizer *synchronizer = Synchronizer::get(); // need to init sync here to break circular dependency...
//writeLock(filename); ScopedWriteLock lock(this, filename);
MetadataFile metadata = MetadataFile(filename); MetadataFile metadata = MetadataFile(filename);
@@ -384,65 +376,71 @@ int IOCoordinator::append(const char *filename, const uint8_t *data, size_t leng
{ {
std::vector<metadataObject>::const_iterator i = objects.begin(); std::vector<metadataObject>::const_iterator i = objects.begin();
// figure out how much data to write to this object if ((objectSize - i->length) > 0) // if this is zero then we can't put anything else in this object
if (offset > i->offset)
{ {
journalOffset = offset - i->offset; // figure out how much data to write to this object
writeLength = min((objectSize - journalOffset),dataRemaining); writeLength = min((objectSize - i->length),dataRemaining);
}
err = replicator->addJournalEntry(i->key.c_str(),&data[count],journalOffset,writeLength); cache->makeSpace(writeLength+JOURNAL_ENTRY_HEADER_SIZE);
err = replicator->addJournalEntry(i->key.c_str(),&data[count],i->length,writeLength);
if (err <= 0) if (err <= 0)
{ {
if ((count + journalOffset) > i->length) metadata.updateEntryLength(i->offset, (count + i->length));
metadata.updateEntryLength(i->offset, (count + journalOffset));
metadata.writeMetadata(filename); metadata.writeMetadata(filename);
logger->log(LOG_ERR,"IOCoordinator::write(): newObject failed to complete write, %u of %u bytes written.",count,length); logger->log(LOG_ERR,"IOCoordinator::append(): journal failed to complete write, %u of %u bytes written.",count,length);
return count; return count;
} }
if ((writeLength + journalOffset) > i->length) metadata.updateEntryLength(i->offset, (writeLength + i->length));
metadata.updateEntryLength(i->offset, (writeLength + journalOffset));
count += err; cache->newJournalEntry(writeLength+JOURNAL_ENTRY_HEADER_SIZE);
dataRemaining -= err;
//cache->makeSpace(journal_data_size) synchronizer->newJournalEntry(i->key);
//Synchronizer::newJournalData(journal_file);
count += writeLength;
dataRemaining -= writeLength;
} }
else }
else if (objects.size() > 1)
{ {
//Something went wrong this shouldn't overlap objects //Something went wrong this shouldn't overlap objects
logger->log(LOG_ERR,"IOCoordinator::append(): overlapping objects found on append.",count,length);
return -1;
} }
// there is no overlapping data, or data goes beyond end of last object // append is starting or adding to a new object
while (dataRemaining > 0 && err >= 0) while (dataRemaining > 0)
{ {
//add a new metaDataObject //add a new metaDataObject
writeLength = min(objectSize,dataRemaining); writeLength = min(objectSize,dataRemaining);
if (count == 0)
{ cache->makeSpace(writeLength);
//this is append and it starting beyond last object in metadata
//figure out if the offset is in this object
}
//cache->makeSpace(size)
// add a new metadata object, this will get a new objectKey NOTE: probably needs offset too // add a new metadata object, this will get a new objectKey NOTE: probably needs offset too
metadataObject newObject = metadata.addMetadataObject(filename,writeLength); metadataObject newObject = metadata.addMetadataObject(filename,writeLength);
// write the new object // write the new object
err = replicator->newObject(newObject.key.c_str(),data,journalOffset,writeLength); err = replicator->newObject(newObject.key.c_str(),data,0,writeLength);
if (err <= 0) if (err <= 0)
{ {
// update metadataObject length to reflect what awas actually written // update metadataObject length to reflect what awas actually written
if ((count + journalOffset) > newObject.length) metadata.updateEntryLength(newObject.offset, (count));
metadata.updateEntryLength(newObject.offset, (count + journalOffset));
metadata.writeMetadata(filename); metadata.writeMetadata(filename);
logger->log(LOG_ERR,"IOCoordinator::write(): newObject failed to complete write, %u of %u bytes written.",count,length); logger->log(LOG_ERR,"IOCoordinator::append(): newObject failed to complete write, %u of %u bytes written.",count,length);
return count; return count;
//log error and abort //log error and abort
} }
// sync cache->newObject(newObject.key,writeLength);
//Synchronizer::newObject(newname) newObjectKeys.push_back(newObject.key);
count += err;
dataRemaining -= err; count += writeLength;
dataRemaining -= writeLength;
} }
synchronizer->newObjects(newObjectKeys);
metadata.writeMetadata(filename); metadata.writeMetadata(filename);
lock.unlock();
return count; return count;
} }

48
src/unit_tests.cpp
View File

@@ -287,6 +287,43 @@ void metadataJournalTest(std::size_t size, off_t offset)
assert(resp->returnCode == (int) size); assert(resp->returnCode == (int) size);
} }
void metadataJournalTest_append(std::size_t size)
{
// make an empty file to write to
const char *filename = "metadataJournalTest";
uint8_t buf[(sizeof(write_cmd)+std::strlen(filename)+size)];
uint64_t *data;
sm_msg_header *hdr = (sm_msg_header *) buf;
append_cmd *cmd = (append_cmd *) &hdr[1];
cmd->opcode = APPEND;
cmd->count = size;
cmd->flen = std::strlen(filename);
memcpy(&cmd->filename, filename, cmd->flen);
data = (uint64_t *) &cmd->filename[cmd->flen];
int count = 0;
for (uint64_t i = 0; i < (size/sizeof(uint64_t)); i++)
{
data[i] = i;
count++;
}
hdr->type = SM_MSG_START;
hdr->payloadLen = sizeof(*cmd) + cmd->flen + cmd->count;
AppendTask a(clientSock, hdr->payloadLen);
int err = ::write(sessionSock, cmd, hdr->payloadLen);
a.run();
// verify response
err = ::recv(sessionSock, buf, 1024, MSG_DONTWAIT);
sm_response *resp = (sm_response *) buf;
assert(err == sizeof(*resp));
assert(resp->header.type == SM_MSG_START);
assert(resp->header.payloadLen == 4);
assert(resp->header.flags == 0);
assert(resp->returnCode == (int) size);
}
void metadataJournalTestCleanup(std::size_t size) void metadataJournalTestCleanup(std::size_t size)
{ {
Config* config = Config::get(); Config* config = Config::get();
@@ -1424,8 +1461,15 @@ int main()
//Case 4 write starts object1 ends object3 //Case 4 write starts object1 ends object3
metadataJournalTest((10*sizeKB),(7*sizeKB)); metadataJournalTest((10*sizeKB),(7*sizeKB));
//Case 5 write starts in new object at offset >0 //Case 5 write starts in new object at offset >0
//TODO add zero padding to writes in this scenario
//metadataJournalTest((8*sizeKB),4*sizeKB); //append test
// first one appends file to end of 8K object
metadataJournalTest_append((7*sizeKB));
// this apppends that starts on new object
metadataJournalTest_append((7*sizeKB));
// this starts in one object and crosses into new object
metadataJournalTest_append((7*sizeKB));
//writetask(); //writetask();
//appendtask(); //appendtask();