First cut of improving IO behavior for BRM load/save

versioning/BRM/extentmap.cpp

@@ -1253,14 +1253,22 @@ void ExtentMap::loadVersion4(IDBDataFile* in)
         growEMShmseg(nrows);
     }
 
-    for (int i = 0; i < emNumElements; i++)
+    size_t progress = 0, writeSize = emNumElements * sizeof(EMEntry);
+    int err;
+    char *writePos = (char *) fExtentMap;
+    while (progress < writeSize)
     {
-        if (in->read((char*) &fExtentMap[i], sizeof(EMEntry)) != sizeof(EMEntry))
+        err = in->read(writePos + progress, writeSize - progress);
+        if (err <= 0)
         {
             log_errno("ExtentMap::loadVersion4(): read ");
             throw runtime_error("ExtentMap::loadVersion4(): read failed. Check the error log.");
         }
-
+        progress += (uint) err;
+    }
+    
+    for (int i = 0; i < emNumElements; i++)
+    {
         reserveLBIDRange(fExtentMap[i].range.start, fExtentMap[i].range.size);
 
         //@bug 1911 - verify status value is valid
@@ -1268,6 +1276,7 @@ void ExtentMap::loadVersion4(IDBDataFile* in)
                 fExtentMap[i].status > EXTENTSTATUSMAX)
             fExtentMap[i].status = EXTENTAVAILABLE;
     }
+    
 
     fEMShminfo->currentSize = emNumElements * sizeof(EMEntry);
 
@@ -1328,7 +1337,8 @@ void ExtentMap::load(const string& filename, bool fixFL)
         throw;
     }
 
-    if (IDBPolicy::useHdfs())
+    // XXXPAT: Forcing the IDB path.  Remove the fstream path once we see this works.
+    if (true || IDBPolicy::useHdfs())
     {
         const char* filename_p = filename.c_str();
         scoped_ptr<IDBDataFile>  in(IDBDataFile::open(
@@ -1363,7 +1373,7 @@ void ExtentMap::load(const string& filename, bool fixFL)
             throw;
         }
     }
-    else
+    else   // fstream path to be remove
     {
         ifstream in;
         in.open(filename.c_str(), ios_base::in | ios_base::binary);
@@ -1443,14 +1453,15 @@ void ExtentMap::save(const string& filename)
         throw runtime_error("ExtentMap::save(): got request to save an empty BRM");
     }
 
-    if (IDBPolicy::useHdfs())
+    // XXXPAT: I don't know why there are two options here.  It can just use the IDBDataFile stuff.
+    // Forcing the IDB option to execute for now.  Leaving the old fstream version there in case we find there's 
+    // a case the IDB option doesn't work.
+    if (true || IDBPolicy::useHdfs())
     {
-        utmp = ::umask(0);
         const char* filename_p = filename.c_str();
         scoped_ptr<IDBDataFile> out(IDBDataFile::open(
                                         IDBPolicy::getType(filename_p, IDBPolicy::WRITEENG),
                                         filename_p, "wb", IDBDataFile::USE_VBUF));
-        ::umask(utmp);
 
         if (!out)
         {
@@ -1484,50 +1495,68 @@ void ExtentMap::save(const string& filename)
         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)
+            if (fExtentMap[i].range.size > 0 && first == -1)
+                first = i;
+            else if (fExtentMap[i].range.size <= 0 && first != -1)
             {
-                try
+                last = i;
+                writeSize = (last - first) * emEntrySize;
+                progress = 0;
+                char *writePos = (char *) &fExtentMap[first];
+                while (progress < writeSize)
                 {
-                    bytes = out->write((char*) &fExtentMap[i], emEntrySize);
+                    err = out->write(writePos + progress, writeSize - progress);
-
+                    if (err < 0)
-                    if (bytes != emEntrySize)
+                    {
+                        releaseFreeList(READ);
+                        releaseEMEntryTable(READ);
                         throw ios_base::failure("ExtentMap::save(): write failed. Check the error log.");
+                    }
+                    progress += err;
                 }
-                catch (...)
+                first = -1;
+            }
+        }
+        if (first != -1)
+        {
+            writeSize = (allocdSize - first) * emEntrySize;
+            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;
+                    throw ios_base::failure("ExtentMap::save(): write failed. Check the error log.");
                 }
+                progress += err;
             }
         }
 
         allocdSize = fFLShminfo->allocdSize / sizeof(InlineLBIDRange);
-        const int inlineLbidRangeSize = sizeof(InlineLBIDRange);
+        //const int inlineLbidRangeSize = sizeof(InlineLBIDRange);
 
-        for (i = 0; i < allocdSize; i++)
+        progress = 0;
+        char *writePos = (char *) fFreeList;
+        while (progress < (size_t) allocdSize)
         {
-//			if (fFreeList[i].size > 0) {
+            err = out->write(writePos + progress, allocdSize - progress);
-            try
+            if (err < 0)
-            {
-                int bytes = out->write((char*) &fFreeList[i], inlineLbidRangeSize);
-
-                if (bytes != inlineLbidRangeSize)
-                    throw ios_base::failure("ExtentMap::save(): write failed. Check the error log.");
-            }
-            catch (...)
             {
                 releaseFreeList(READ);
                 releaseEMEntryTable(READ);
-                throw;
+                throw ios_base::failure("ExtentMap::save(): write failed. Check the error log.");
             }
-
+            progress += err;
-//			}
         }
     }
-    else
+    else    // this is the fstream version to be expired
     {
         ofstream out;
 

versioning/BRM/oidserver.cpp

@@ -349,12 +349,17 @@ void OIDServer::initializeBitmap() const
     writeData(buf, 0, HeaderSize);
 
     // reset buf to all 0's and write the bitmap
-    for (i = 0; i < HeaderSize; i++)
+    //for (i = 0; i < HeaderSize; i++)
-        buf[i] = 0;
+    //    buf[i] = 0;
 
-    for (i = 0; i < bitmapSize; i += HeaderSize)
+    //for (i = 0; i < bitmapSize; i += HeaderSize)
-        writeData(buf, HeaderSize + i, (bitmapSize - i > HeaderSize ? HeaderSize : bitmapSize - i));
+    //    writeData(buf, HeaderSize + i, (bitmapSize - i > HeaderSize ? HeaderSize : bitmapSize - i));
 
+    uint8_t *bitmapbuf = new uint8_t[bitmapSize];
+    memset(bitmapbuf, 0, bitmapSize);
+    writeData(bitmapbuf, HeaderSize, bitmapSize);
+    delete bitmapbuf;
+    
     flipOIDBlock(0, firstOID, 0);
 
     /* append a 16-bit 0 to indicate 0 entries in the vboid->dbroot mapping */

versioning/BRM/vbbm.cpp

@@ -1067,6 +1067,32 @@ void VBBM::loadVersion2(IDBDataFile* in)
         throw runtime_error("VBBM::load(): Failed to load vb file meta data");
     }
 
+    size_t readSize = vbbmEntries * sizeof(entry);
+    char *readBuf = new char[readSize];
+    size_t progress = 0;
+    int err;
+    while (progress < readSize)
+    {
+        err = in->read(readBuf + progress, readSize - progress);
+        if (err < 0)
+        {
+            log_errno("VBBM::load()");
+            throw runtime_error("VBBM::load(): Failed to load, check the critical log file");
+        }
+        else if (err == 0)
+        {
+            log("VBBM::load(): Got early EOF");
+            throw runtime_error("VBBM::load(): Got early EOF");
+        }
+        progress += err;
+    }
+    
+    VBBMEntry *loadedEntries = (VBBMEntry *) readBuf;
+    for (i = 0; i < vbbmEntries; i++)
+        insert(loadedEntries[i].lbid, loadedEntries[i].verID, loadedEntries[i].vbOID,
+          loadedEntries[i].vbFBO, true);
+    
+    /*
     for (i = 0; i < vbbmEntries; i++)
     {
         if (in->read((char*)&entry, sizeof(entry)) != sizeof(entry))
@@ -1077,6 +1103,7 @@ void VBBM::loadVersion2(IDBDataFile* in)
 
         insert(entry.lbid, entry.verID, entry.vbOID, entry.vbFBO, true);
     }
+    */
 
 }
 
@@ -1132,16 +1159,15 @@ void VBBM::load(string filename)
 void VBBM::save(string filename)
 {
     int i;
-    mode_t utmp = ::umask(0);
     int var;
 
-    if (IDBPolicy::useHdfs())
+    // XXXPAT: forcing the IDB* path.  Delete the fstream path when appropriate.
+    if (true || IDBPolicy::useHdfs())
     {
         const char* filename_p = filename.c_str();
         scoped_ptr<IDBDataFile> out(IDBDataFile::open(
                                         IDBPolicy::getType(filename_p, IDBPolicy::WRITEENG),
                                         filename_p, "wb", IDBDataFile::USE_VBUF));
-        ::umask(utmp);
 
         if (!out)
         {
@@ -1159,6 +1185,50 @@ void VBBM::save(string filename)
         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;
+            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;
+            }
+        }
+
+        /*
         for (i = 0; i < vbbm->vbCapacity; i++)
         {
             if (storage[i].lbid != -1)
@@ -1173,12 +1243,13 @@ void VBBM::save(string filename)
             log_errno("VBBM::save()");
             throw runtime_error("VBBM::save(): Failed to write the file");
         }
+        */
     }
     else
     {
         ofstream out;
         out.open(filename.c_str(), ios_base::trunc | ios_base::out | ios_base::binary);
-        ::umask(utmp);
+        //::umask(utmp);
 
         if (!out)
         {

versioning/BRM/vss.cpp

@@ -1293,15 +1293,14 @@ void VSS::save(string filename)
 {
     int i;
     struct Header header;
-    mode_t utmp = ::umask(0);
 
-    if (IDBPolicy::useHdfs())
+    // XXXPAT: Forcing the IDB* path to run.  Delete the fstream path when appropriate.
+    if (true || IDBPolicy::useHdfs())
     {
         const char* filename_p = filename.c_str();
         scoped_ptr<IDBDataFile> out(IDBDataFile::open(
                                         IDBPolicy::getType(filename_p, IDBPolicy::WRITEENG),
                                         filename_p, "wb", IDBDataFile::USE_VBUF));
-        ::umask(utmp);
 
         if (!out)
         {
@@ -1318,6 +1317,49 @@ void VSS::save(string filename)
             throw runtime_error("VSS::save(): Failed to write header to the file");
         }
 
+        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;
+            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;
+            }
+        }
+        
+        /*
         for (i = 0; i < vss->capacity; i++)
         {
             if (storage[i].lbid != -1)
@@ -1329,12 +1371,13 @@ void VSS::save(string filename)
                 }
             }
         }
+        */
     }
     else
     {
         ofstream out;
         out.open(filename.c_str(), ios_base::trunc | ios_base::out | ios_base::binary);
-        ::umask(utmp);
+        //::umask(utmp);
 
         if (!out)
         {
@@ -1450,6 +1493,32 @@ void VSS::load(string filename)
     	vss->LWM = 0;
     */
 
+    size_t readSize = header.entries * sizeof(entry);
+    char *readBuf = new char[readSize];
+    size_t progress = 0;
+    int err;
+    while (progress < readSize)
+    {
+        err = in->read(readBuf + progress, readSize - progress);
+        if (err < 0)
+        {
+            log_errno("VBBM::load()");
+            throw runtime_error("VBBM::load(): Failed to load, check the critical log file");
+        }
+        else if (err == 0)
+        {
+            log("VBBM::load(): Got early EOF");
+            throw runtime_error("VBBM::load(): Got early EOF");
+        }
+        progress += err;
+    }
+    
+    VSSEntry *loadedEntries = (VSSEntry *) readBuf;
+    for (i = 0; i < header.entries; i++)
+        insert(loadedEntries[i].lbid, loadedEntries[i].verID, loadedEntries[i].vbFlag, 
+          loadedEntries[i].locked, true);
+
+    /*
     for (i = 0; i < header.entries; i++)
     {
         if (in->read((char*)&entry, sizeof(entry)) != sizeof(entry))
@@ -1460,7 +1529,8 @@ void VSS::load(string filename)
 
         insert(entry.lbid, entry.verID, entry.vbFlag, entry.locked, true);
     }
-
+    */
+    
     //time2 = microsec_clock::local_time();
     //cout << "done loading " << time2 << " duration: " << time2-time1 << endl;
 }