ryzom/server/src/pd_lib/pds_table_buffer.cpp

   1 // Ryzom - MMORPG Framework <http://dev.ryzom.com/projects/ryzom/>
   2 // Copyright (C) 2010  Winch Gate Property Limited
   3 //
   4 // This program is free software: you can redistribute it and/or modify
   5 // it under the terms of the GNU Affero General Public License as
   6 // published by the Free Software Foundation, either version 3 of the
   7 // License, or (at your option) any later version.
   8 //
   9 // This program is distributed in the hope that it will be useful,
  10 // but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 // GNU Affero General Public License for more details.
  13 //
  14 // You should have received a copy of the GNU Affero General Public License
  15 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16
  17 #include "pds_table_buffer.h"
  18
  19 #include "nel/misc/time_nl.h"
  20
  21 #include "pd_lib.h"
  22 #include "pd_utils.h"
  23
  24 #include <time.h>
  25
  26 uint32  CTableBuffer::_MaxRefFileSize = 32768; //128*1024*1024;
  27
  28 // Common stamp
  29 uint32  CTableBuffer::_CommonStamp = 0;
  30
  31 /*
  32  * Constructor
  33  */
  34 CTableBuffer::CTableBuffer()
  35 {
  36         clear();
  37 }
  38
  39 /*
  40  * Destructor
  41  */
  42 CTableBuffer::~CTableBuffer()
  43 {
  44         clear();
  45 }
  46
  47 /*
  48  * Clear all
  49  */
  50 void    CTableBuffer::clear()
  51 {
  52         _Init = false;
  53         _TableId = 0;
  54         _RowSize = 0;
  55         _InternalRowSize = 0;
  56         _RowsPerFile = 0;
  57         _RowMap.clear();
  58         _ReleaseSet.clear();
  59         _DirtyList.clear();
  60         purgeReferences();
  61         _RowMapper.clear();
  62         _Mapped = false;
  63         _ReferenceStamp = 0;
  64         _CurrentDeltaId = 0;
  65 }
  66
  67 /*
  68  * Init
  69  */
  70 void    CTableBuffer::init(uint32 tableId, uint32 rowSize, bool mapped)
  71 {
  72         clear();
  73
  74         _TableId = tableId;
  75         _RowSize = rowSize;
  76         _Mapped = mapped;
  77         _InternalRowSize = _RowSize + getHeaderSize();
  78         // compute maximum number of rows a ref file will contain
  79         if (!initRowsPerFile())
  80                 return;
  81
  82         _Init = true;
  83 }
  84
  85 /*
  86  * Init
  87  */
  88 void    CTableBuffer::init(uint32 tableId, const std::string& refRootPath, const std::string& refPath) //CRefIndex& ref)
  89 {
  90         clear();
  91
  92         _TableId = tableId;
  93
  94         _RefRootPath = refRootPath;
  95         _RefPath = refPath;
  96
  97         _Init = true;
  98 }
  99
 100 /*
 101  * Setup Ref
 102  */
 103 void    CTableBuffer::setupRef(CRefIndex& ref)
 104 {
 105         nlassert(_Init);
 106
 107         _RefRootPath = ref.getRootPath();
 108         _RefPath = ref.getPath();
 109         _ReferenceStamp = uint32(ref.Timestamp.toTime());
 110
 111         // clear all ref files
 112         purgeReferences();
 113
 114         // release all rows that can be released
 115         flushReleased();
 116 }
 117
 118
 119
 120
 121 /*
 122  * Return a pointer to straight row data
 123  * If row is not loaded, data are loaded from table file
 124  */
 125 CTableBuffer::CAccessor CTableBuffer::getRow(RY_PDS::TRowIndex row)
 126 {
 127         nlassert(_Init);
 128
 129         TRowData                        rowData = NULL;
 130         TRowMap::iterator       it = _RowMap.find(row);
 131
 132         PDS_FULL_DEBUG("getRow(): row '%d'", row);
 133
 134         if (it == _RowMap.end())
 135         {
 136                 rowData = new uint8[_InternalRowSize];
 137
 138                 it = _RowMap.insert(TRowMap::value_type(row, rowData)).first;
 139                 loadRow(row, rowData);
 140         }
 141
 142         acquireRow(it);
 143
 144         return CAccessor(it, _Mapped);
 145 }
 146
 147 /*
 148  * Return a pointer to straight new row data
 149  * Row is filled with blank, not loaded from file
 150  */
 151 CTableBuffer::CAccessor CTableBuffer::getEmptyRow(RY_PDS::TRowIndex row)
 152 {
 153         nlassert(_Init);
 154
 155         TRowData                        rowData = NULL;
 156         TRowMap::iterator       it = _RowMap.find(row);
 157
 158         if (it == _RowMap.end())
 159         {
 160                 rowData = new uint8[_InternalRowSize];
 161                 it = _RowMap.insert(TRowMap::value_type(row, rowData)).first;
 162                 memset(rowData, 0, _InternalRowSize);
 163         }
 164         else
 165         {
 166                 PDS_WARNING("getEmptyRow(): row '%d' already exists! data might be lost if buffer modified!", row);
 167         }
 168
 169         acquireRow(it);
 170
 171         return CAccessor(it, _Mapped);
 172 }
 173
 174 /*
 175  * Acquire a Row
 176  * Row is marked as non releasable until it is released enough time
 177  */
 178 bool    CTableBuffer::acquireRow(CAccessor accessor)
 179 {
 180         nlassert(_Init);
 181
 182         return acquireRow(accessor._MapIt);
 183 }
 184
 185 /*
 186  * Release a Row
 187  * Row is marked as purgeable, and will be purged as soon as possible
 188  * unless it is reactivated before purge
 189  */
 190 bool    CTableBuffer::releaseRow(CAccessor accessor)
 191 {
 192         nlassert(_Init);
 193
 194         return releaseRow(accessor._MapIt);
 195 }
 196
 197 /*
 198  * Release a Row
 199  * Row is marked as purgeable, and will be purged as soon as possible
 200  * unless it is reactivated before purge
 201  */
 202 bool    CTableBuffer::releaseRow(RY_PDS::TRowIndex row)
 203 {
 204         nlassert(_Init);
 205
 206         TRowMap::iterator       it = _RowMap.find(row);
 207
 208         if (it == _RowMap.end())
 209         {
 210                 return false;
 211         }
 212
 213         return releaseRow(it);
 214 }
 215
 216 /*
 217  * Release all rows
 218  */
 219 bool    CTableBuffer::releaseAll()
 220 {
 221         TRowMap::iterator       it, itr;
 222
 223         for (it=_RowMap.begin(); it!=_RowMap.end(); )
 224         {
 225                 // force all rows as unacquired
 226                 itr = it++;
 227                 releaseRow(itr, true);
 228         }
 229
 230         return true;
 231 }
 232
 233
 234
 235
 236
 237
 238 /*
 239  * Load a row from the appropriate reference file
 240  */
 241 bool    CTableBuffer::loadRow(RY_PDS::TRowIndex row, TRowData rowData)
 242 {
 243         nlassert(_Init);
 244
 245         // locate reference file
 246         uint32  file = row / _RowsPerFile;
 247
 248         // check reference is ready
 249         if (!checkRef(file))
 250                 return false;
 251
 252         CDBReferenceFile&       refFile = *(_RefFileMap[file]);
 253
 254         if (!refFile.read(row, rowData))
 255                 return false;
 256
 257         // row comes from reference
 258         // clear Dirty and AcquireCount, as row was loaded from reference
 259         CHeader*        hdr = (CHeader*)rowData;
 260
 261         hdr->clearDirtStamp();
 262         hdr->clearAcquireCount();
 263
 264         return true;
 265 }
 266
 267
 268
 269 /*
 270  * Update a Row
 271  * Update in file the whole row
 272  */
 273 bool    CTableBuffer::updateRow(CAccessor accessor)
 274 {
 275         return updateRow(accessor.row(), accessor.fullRow(), true);
 276 }
 277
 278 /*
 279  * Update a Row
 280  */
 281 bool    CTableBuffer::updateRow(RY_PDS::TRowIndex row, const TRowData rowData, bool forceWriteToDisk)
 282 {
 283         nlassert(_Init);
 284
 285         TRowMap::iterator       it = _RowMap.find(row);
 286
 287         // if row is not mapped in ram or if disk write is forced
 288         if (it == _RowMap.end() || forceWriteToDisk)
 289         {
 290                 // locate reference file
 291                 uint32  file = row / _RowsPerFile;
 292                 // check reference is ready
 293                 if (!checkRef(file))
 294                         return false;
 295
 296                 // update row
 297                 CDBReferenceFile&       refFile = *(_RefFileMap[file]);
 298                 if (!refFile.update(row, rowData))
 299                         return false;
 300         }
 301
 302         // if row is mapped in ram
 303         if (it != _RowMap.end())
 304         {
 305                 // get row data buffer
 306                 TRowData        dest = (*it).second;
 307
 308                 // row is clean and warm from reference file
 309                 ((CHeader*)rowData)->clearDirtStamp();
 310
 311                 // copy data
 312                 memcpy(dest, rowData, _InternalRowSize);
 313         }
 314
 315         return true;
 316 }
 317
 318 /*
 319  * Mark a Row as being dirty for later delta save
 320  */
 321 bool    CTableBuffer::dirtyRow(CAccessor accessor)
 322 {
 323         nlassert(_Init);
 324
 325         CHeader*        header = (CHeader*)accessor.fullRow();
 326
 327         // check row not dirty already
 328         if (!header->dirty())
 329         {
 330                 // add row to list
 331                 header->setDirty();
 332                 _DirtyList.push_back(accessor);
 333         }
 334
 335         return true;
 336 }
 337
 338 /*
 339  * Build Delta
 340  */
 341 bool    CTableBuffer::buildDelta(const CTimestamp& starttimestamp, const CTimestamp& endtimestamp)
 342 {
 343         nlassert(_Init);
 344
 345         // check there is something to write
 346         if (_DirtyList.empty())
 347                 return true;
 348
 349         // setup delta file
 350         CDBDeltaFile    delta;
 351         std::string             deltaFilename = CDBDeltaFile::getDeltaFileName(_TableId, endtimestamp);
 352         delta.setup(deltaFilename, _RefRootPath+"seconds", _InternalRowSize, starttimestamp, endtimestamp);
 353
 354         // setup delta id
 355         delta.setDeltaIds(_CurrentDeltaId, _CurrentDeltaId);
 356         ++_CurrentDeltaId;
 357
 358         uint    i;
 359         // go through all dirty rows
 360         for (i=0; i<_DirtyList.size(); ++i)
 361         {
 362                 RY_PDS::TRowIndex       row = _DirtyList[i].row();
 363                 TRowData                        data = _DirtyList[i].fullRow();
 364
 365                 // clean dirty flag
 366                 ((CHeader*)data)->clearFlags(CHeader::Dirty);
 367
 368                 if (!delta.write(row, data))
 369                         return false;
 370         }
 371
 372         // clear list
 373         _DirtyList.clear();
 374
 375         //
 376         std::string             filename = _RefRootPath+"seconds/"+deltaFilename;
 377         PDS_DEBUG("buildDelta(): built table delta '%s', %d rows written", filename.c_str(), i);
 378
 379         return true;
 380 }
 381
 382 /*
 383  * Apply delta changes from a file
 384  */
 385 bool    CTableBuffer::applyDeltaChanges(const std::string& filename)
 386 {
 387         CDBDeltaFile    delta;
 388
 389         delta.setup(filename, _InternalRowSize, CTimestamp(), CTimestamp());
 390
 391         if (!delta.preload())
 392         {
 393                 PDS_WARNING("applyDeltaChanges(): failed to preload file '%s'", filename.c_str());
 394                 return false;
 395         }
 396
 397         // internal row size not set, get it from delta file
 398         if (_InternalRowSize == 0)
 399         {
 400                 // get row size from delta
 401                 _InternalRowSize = delta.getRowSize();
 402                 // recompute good number of rows per file
 403                 initRowsPerFile();
 404         }
 405         else if (_InternalRowSize != delta.getRowSize())
 406         {
 407                 PDS_WARNING("applyDeltaChanges(): delta file '%s' has mismatching row size (%d bytes expected, %d bytes found)", filename.c_str(), _InternalRowSize, delta.getRowSize());
 408                 return false;
 409         }
 410
 411         uint32                          startDeltaId, endDeltaId;
 412         delta.getDeltaIds(startDeltaId, endDeltaId);
 413
 414         if (!updateDeltaIds(startDeltaId, endDeltaId))
 415         {
 416                 PDS_WARNING("applyDeltaChanges(): failed to update delta ids from file '%s'", filename.c_str());
 417                 return false;
 418         }
 419
 420         uint32                          index;
 421         std::vector<uint8>      buffer(_InternalRowSize);
 422         uint8*                          data = &(buffer[0]);
 423
 424         // read data from delta till the end
 425         while (true)
 426         {
 427                 // read data from delta file
 428                 if (!delta.read(index, data))
 429                 {
 430                         PDS_WARNING("applyDeltaChanges(): failed to load next row from delta file '%s'", filename.c_str());
 431                         return false;
 432                 }
 433
 434                 if (index == 0xffffffff)
 435                         return true;
 436
 437                 // update disk data with read buffer
 438                 if (!updateRow(index, data, true))
 439                 {
 440                         PDS_WARNING("applyDeltaChanges(): failed to update row '%d' data from file '%s'", index, filename.c_str());
 441                         return false;
 442                 }
 443         }
 444
 445         return true;
 446 }
 447
 448
 449 /*
 450  * Flush Released Rows from memory
 451  */
 452 void    CTableBuffer::flushReleased()
 453 {
 454         TReleaseSet::iterator   it;
 455
 456         // go through all released rows
 457         for (it=_ReleaseSet.begin(); it!=_ReleaseSet.end(); )
 458         {
 459                 // get row
 460                 TRowMap::iterator       rit = _RowMap.find(*it);
 461                 if (rit == _RowMap.end())
 462                 {
 463                         PDS_WARNING("flushReleased(): row '%d' not present, already released?", *it);
 464                         // anyway, remove from released set
 465                         TReleaseSet::iterator   itr = (it++);
 466                         _ReleaseSet.erase(itr);
 467                         continue;
 468                 }
 469
 470                 // check row has really been released
 471                 CHeader*        header = (CHeader*)(*rit).second;
 472                 if (!header->isAcquired())
 473                 {
 474                         PDS_WARNING("flushReleased(): try to release row '%d' not flagged as being released, bypassed", *it);
 475                         // remove from release set
 476                         TReleaseSet::iterator   itr = (it++);
 477                         _ReleaseSet.erase(itr);
 478                         continue;
 479                 }
 480
 481                 // the row may have been released and dirtied at the same time
 482                 // don't flush it, since it has not been deltaed...
 483                 if (header->dirty() || header->getDirtStamp() >= _ReferenceStamp)
 484                 {
 485                         ++it;
 486                         continue;
 487                 }
 488
 489                 // delete row
 490                 delete (*rit).second;
 491                 _RowMap.erase(rit);
 492
 493                 // remove from set
 494                 TReleaseSet::iterator   itr = (it++);
 495                 _ReleaseSet.erase(itr);
 496         }
 497 }
 498
 499 /*
 500  * Reset dirty tags
 501  */
 502 void    CTableBuffer::resetDirty()
 503 {
 504         _DirtyList.clear();
 505
 506         TRowMap::iterator       it;
 507         for (it=_RowMap.begin(); it!=_RowMap.end(); ++it)
 508         {
 509                 CHeader*        header = (CHeader*)((*it).second);
 510                 header->clearFlags(CHeader::Dirty);
 511         }
 512 }
 513
 514 /*
 515  * Flush Reference files
 516  */
 517 void    CTableBuffer::flushRefFiles()
 518 {
 519         uint    i;
 520         for (i=0; i<_RefFileMap.size(); ++i)
 521                 if (_RefFileMap[i] != NULL && _RefFileMap[i]->initialised())
 522                         _RefFileMap[i]->flush();
 523 }
 524
 525 /*
 526  * Purge all references
 527  */
 528 bool    CTableBuffer::purgeReferences()
 529 {
 530         uint    i;
 531         for (i=0; i<_RefFileMap.size(); ++i)
 532         {
 533                 if (_RefFileMap[i] != NULL && _RefFileMap[i]->initialised())
 534                 {
 535                         _RefFileMap[i]->flush();
 536                         delete _RefFileMap[i];
 537                         _RefFileMap[i] = NULL;
 538                 }
 539         }
 540
 541         _RefFileMap.clear();
 542
 543         return true;
 544 }
 545
 546 /*
 547  * Open all reference files in reference directory
 548  */
 549 bool    CTableBuffer::openAllRefFilesWrite()
 550 {
 551         std::vector<std::string>        refs;
 552
 553         getReferenceFilesList(refs);
 554
 555         uint    i;
 556         for (i=0; i<refs.size(); ++i)
 557         {
 558                 uint32  tableId;
 559                 uint32  refFileId;
 560
 561                 // check file is a reference of the table
 562                 if (!CDBReferenceFile::isRefFile(refs[i], tableId, refFileId) || tableId != _TableId)
 563                 {
 564                         PDS_WARNING("openAllRefFilesWrite(): file '%s' is not a reference file for table '%d'", refs[i].c_str(), _TableId);
 565                         continue;
 566                 }
 567
 568                 if (!checkRef(refFileId))
 569                 {
 570                         PDS_WARNING("openAllRefFilesWrite(): failed to check reference file '%s' for table '%d'", refs[i].c_str(), _TableId);
 571                         return false;
 572                 }
 573
 574                 if (!_RefFileMap[refFileId]->prewrite())
 575                 {
 576                         PDS_WARNING("openAllRefFilesWrite(): failed to check prewrite reference file '%s' for table '%d'", refs[i].c_str(), _TableId);
 577                         return false;
 578                 }
 579         }
 580
 581         return true;
 582 }
 583
 584 /*
 585  * Open all reference files in reference directory
 586  */
 587 bool    CTableBuffer::openAllRefFilesRead()
 588 {
 589         std::vector<std::string>        refs;
 590
 591         getReferenceFilesList(refs);
 592
 593         uint32  startId = 0xffffffff;
 594         uint32  endId = 0xffffffff;
 595
 596         uint    i;
 597         for (i=0; i<refs.size(); ++i)
 598         {
 599                 uint32  tableId;
 600                 uint32  refFileId;
 601
 602                 // check file is a reference of the table
 603                 if (!CDBReferenceFile::isRefFile(refs[i], tableId, refFileId) || tableId != _TableId)
 604                 {
 605                         PDS_WARNING("openAllRefFilesRead(): file '%s' is not a reference file for table '%d'", refs[i].c_str(), _TableId);
 606                         continue;
 607                 }
 608
 609                 if (!checkRef(refFileId))
 610                 {
 611                         PDS_WARNING("openAllRefFilesRead(): failed to check reference file '%s' for table '%d'", refs[i].c_str(), _TableId);
 612                         return false;
 613                 }
 614
 615                 if (!_RefFileMap[refFileId]->preload())
 616                 {
 617                         PDS_WARNING("openAllRefFilesRead(): failed to check preload reference file '%s' for table '%d'", refs[i].c_str(), _TableId);
 618                         return false;
 619                 }
 620
 621                 uint32  endIdCheck;
 622                 _RefFileMap[refFileId]->getUpdateDeltaIds(startId, endIdCheck);
 623
 624                 if (endId == 0xffffffff)
 625                 {
 626                         endId = endIdCheck;
 627                 }
 628                 else if (endId != endIdCheck)
 629                 {
 630                         PDS_WARNING("openAllRefFilesRead(): expected endId '%d', found '%d' in file '%s'", endId, endIdCheck, refs[i].c_str());
 631                         return false;
 632                 }
 633         }
 634
 635         _CurrentDeltaId = endId + 1;
 636
 637         return true;
 638 }
 639
 640
 641 /*
 642  * Update delta ids for all references
 643  */
 644 bool    CTableBuffer::updateDeltaIds(uint32 start, uint32 end)
 645 {
 646         uint    i;
 647         for (i=0; i<_RefFileMap.size(); ++i)
 648         {
 649                 if (_RefFileMap[i] == NULL || !_RefFileMap[i]->initialised())
 650                 {
 651                         PDS_WARNING("updateDeltaIds(): for table '%d', file %d not initialised", _TableId, i);
 652                         return false;
 653                 }
 654
 655                 if (!_RefFileMap[i]->updateDeltaIds(start, end))
 656                 {
 657                         PDS_WARNING("updateDeltaIds(): failed to update delta ids for table %d, file %d", _TableId, i);
 658                         return false;
 659                 }
 660         }
 661
 662         return true;
 663 }
 664
 665
 666
 667 /*
 668  * Get Reference Files list
 669  */
 670 void    CTableBuffer::getReferenceFilesList(std::vector<std::string> &result)
 671 {
 672         result.clear();
 673         std::vector<std::string>        files;
 674         NLMISC::CPath::getPathContent(_RefPath, false, false, true, files);
 675
 676         uint    i;
 677         // check all files in reference directory
 678         for (i=0; i<files.size(); ++i)
 679         {
 680                 uint32  tableId;
 681                 uint32  refFileId;
 682
 683                 // check file is a reference of the table
 684                 if (!CDBReferenceFile::isRefFile(files[i], tableId, refFileId) || tableId != _TableId)
 685                         continue;
 686
 687                 if (result.size() <= refFileId)
 688                         result.resize(refFileId+1);
 689
 690                 result[refFileId] = files[i];
 691         }
 692 }
 693
 694
 695 /*
 696  * Build RowMapper
 697  */
 698 bool    CTableBuffer::buildRowMapper()
 699 {
 700         std::vector<std::string>        files;
 701         getReferenceFilesList(files);
 702
 703         std::vector<uint8>      buffer(_InternalRowSize);
 704         uint8*  rowData = &(buffer[0]);
 705
 706         _RowMapper.clear();
 707
 708         uint    i;
 709         // check all files in reference directory
 710         for (i=0; i<files.size(); ++i)
 711         {
 712                 if (files[i].empty())
 713                         continue;
 714
 715                 // get row indices in reference file
 716                 RY_PDS::TRowIndex       base, end, row;
 717
 718                 CDBReferenceFile        refFile;
 719
 720                 // init and preload file
 721                 refFile.setup(NLMISC::CFile::getFilename(files[i]), _RefPath, 0, 0, _InternalRowSize);
 722                 if (!refFile.preload())
 723                 {
 724                         PDS_WARNING("buildRowMapper(): failed to preload() '%s'", files[i].c_str());
 725                         return false;
 726                 }
 727
 728                 // get row indices in reference file
 729                 base = refFile.getBaseIndex();
 730                 end = refFile.getEndIndex();
 731
 732                 refFile.close();
 733
 734                 for (row=base; row<end; ++row)
 735                 {
 736                         CAccessor               accessor = getRow(row);
 737
 738                         if (!processRow(accessor))
 739                         {
 740                                 PDS_WARNING("buildRowMapper(): failed to process row '%d' in file '%s'", row, files[i].c_str());
 741                                 releaseRow(accessor);
 742                                 return false;
 743                         }
 744
 745                         releaseRow(accessor);
 746                 }
 747         }
 748
 749         return true;
 750 }
 751
 752 /*
 753  * Process row for RowMapper
 754  */
 755 bool    CTableBuffer::processRow(CAccessor& accessor)
 756 {
 757         CMappedHeader*  header = (CMappedHeader*)accessor.fullRow();
 758
 759         // allocate row if needed
 760         if (!header->allocated())
 761                 return true;
 762
 763         if (!_RowMapper.allocate(accessor.row()))
 764         {
 765                 PDS_WARNING("processRow(): failed to allocate row '%d'", accessor.row());
 766                 return false;
 767         }
 768
 769         // map row if need
 770         // unallocated rows shouldn't be mapped...
 771         if (!_Mapped)
 772                 return true;
 773
 774         if (!header->mapped())
 775         {
 776                 if (!RY_PDS::ResolveUnmappedRows)
 777                 {
 778                         PDS_WARNING("processRow(): failed, row '%d' not mapped", accessor.row());
 779                         return false;
 780                 }
 781                 return true;
 782         }
 783
 784         if (_RowMapper.isMapped(header->getKey()))
 785         {
 786                 // check key not yet mapped
 787                 RY_PDS::CObjectIndex    prevMap = _RowMapper.get(header->getKey());
 788
 789                 // already mapped
 790                 if (!RY_PDS::ResolveDoubleMappedRows)
 791                 {
 792                         PDS_WARNING("processRow(): key '%016" NL_I64 "X' already mapped to '%s', failed", header->getKey(), prevMap.toString().c_str());
 793                         return false;
 794                 }
 795
 796                 PDS_WARNING("processRow(): key '%016" NL_I64 "X' already mapped to '%s'", header->getKey(), prevMap.toString().c_str());
 797
 798                 if (RY_PDS::ResolveDoubleMappedKeepOlder)
 799                 {
 800                         // clear header map
 801                         header->setKey(0);
 802                         dirtyRow(accessor);
 803                         return true;
 804                 }
 805
 806                 unmapRow(prevMap, header->getKey());
 807         }
 808
 809         if (!_RowMapper.map(header->getKey(), RY_PDS::CObjectIndex((RY_PDS::TTableIndex)_TableId, accessor.row())))
 810         {
 811                 PDS_WARNING("processRow(): failed to map row '%d'", accessor.row());
 812                 return false;
 813         }
 814
 815         return true;
 816 }
 817
 818
 819
 820 /*
 821  * Process Rows, apply external row processing after rows are loaded
 822  */
 823 bool    CTableBuffer::processRows(IRowProcessor* processor)
 824 {
 825         RY_PDS::TRowIndex       row;
 826
 827         for (row=0; row<maxRowIndex(); ++row)
 828         {
 829                 // process only allocated rows
 830                 if (!_RowMapper.allocated(row))
 831                         continue;
 832
 833                 CAccessor               accessor = getRow(row);
 834
 835                 if (!processor->processRow((RY_PDS::TTableIndex)_TableId, accessor))
 836                 {
 837                         PDS_WARNING("processRows(): failed to process row '%d'", row);
 838                         releaseRow(accessor);
 839                         return false;
 840                 }
 841
 842                 releaseRow(accessor);
 843         }
 844
 845         return true;
 846 }
 847
 848
 849
 850 /*
 851  * Allocate a row in a table
 852  * \param row is the row to allocate
 853  * Return true if succeeded
 854  */
 855 bool    CTableBuffer::allocate(RY_PDS::TRowIndex row, CAccessor& accessor)
 856 {
 857         // check row is free
 858         if (_RowMapper.allocated(row))
 859         {
 860                 PDS_WARNING("allocate(): row '%d' already allocated", row);
 861                 return false;
 862         }
 863
 864         accessor = getRow(row);
 865         CMappedHeader   *header = (CMappedHeader*)accessor.fullRow();
 866         header->setFlags(CHeader::Allocated);
 867         _RowMapper.allocate(row);
 868
 869         // just in case, clear map key
 870         if (_Mapped && header->getKey() != 0)
 871         {
 872                 header->setKey(0);
 873         }
 874
 875         return dirtyRow(accessor);
 876 }
 877
 878 /*
 879  * Deallocate a row in a table
 880  * \param row is the row to deallocate
 881  * Return true if succeeded
 882  */
 883 bool    CTableBuffer::deallocate(RY_PDS::TRowIndex row)
 884 {
 885         // check row is allocated
 886         if (!_RowMapper.allocated(row))
 887         {
 888                 PDS_WARNING("deallocate(): row '%d' not yet allocated", row);
 889                 return false;
 890         }
 891
 892         CAccessor               accessor = getRow(row);
 893         CMappedHeader   *header = (CMappedHeader*)accessor.fullRow();
 894
 895         // unmap row if was previously mapped -- just in case unmap not called
 896         if (_Mapped && header->getKey() != 0)
 897         {
 898                 _RowMapper.unmap(header->getKey());
 899                 header->setKey(0);
 900         }
 901
 902         header->clearAcquireCount();
 903         header->clearFlags(CHeader::Allocated);
 904         _RowMapper.deallocate(row);
 905
 906         return dirtyRow(accessor);
 907 }
 908
 909 /*
 910  * Map a row
 911  * \param row is the row to allocate
 912  * \param key is the 64 bits row key
 913  * Return true if succeeded
 914  */
 915 bool    CTableBuffer::mapRow(const RY_PDS::CObjectIndex &index, uint64 key)
 916 {
 917         if (!_Mapped)
 918         {
 919                 PDS_WARNING("mapRow(): table not mapped");
 920                 return false;
 921         }
 922
 923         // check row is allocated
 924         if (!_RowMapper.allocated(index.row()))
 925         {
 926                 PDS_WARNING("mapRow(): row '%d' not yet allocated", index.row());
 927                 return false;
 928         }
 929
 930         CAccessor               accessor = getRow(index.row());
 931         CMappedHeader   *header = (CMappedHeader*)accessor.fullRow();
 932
 933         if (!_RowMapper.map(key, index))
 934         {
 935                 PDS_WARNING("mapRow(): failed to map row '%d' to key '%016" NL_I64 "X'", index.row(), key);
 936                 return false;
 937         }
 938
 939         // unmap row if was previously mapped
 940         if (header->getKey() != 0)
 941         {
 942                 _RowMapper.unmap(header->getKey());
 943         }
 944
 945         header->setKey(key);
 946         dirtyRow(accessor);
 947
 948         return true;
 949 }
 950
 951 /*
 952  * Unmap a row in a table
 953  * \param tableIndex is the table to find row
 954  * \param key is the 64 bits row key
 955  * Return true if succeeded
 956  */
 957 bool    CTableBuffer::unmapRow(const RY_PDS::CObjectIndex &index, uint64 key)
 958 {
 959         if (!_Mapped)
 960         {
 961                 PDS_WARNING("unmapRow(): table not mapped");
 962                 return false;
 963         }
 964
 965         // check row is allocated
 966         if (!_RowMapper.allocated(index.row()))
 967         {
 968                 PDS_WARNING("unmapRow(): row '%d' not yet allocated", index.row());
 969                 return false;
 970         }
 971
 972         if (!_RowMapper.unmap(key))
 973         {
 974                 PDS_WARNING("unmapRow(): failed to unmap row '%d' from key '%016" NL_I64 "X", index.row(), key);
 975                 return false;
 976         }
 977
 978         CAccessor               accessor = getRow(index.row());
 979         CMappedHeader   *header = (CMappedHeader*)accessor.fullRow();
 980
 981         if (header->getKey() != key)
 982         {
 983                 PDS_WARNING("unmapRow(): row '%d' is mapped to '%016" NL_I64 "X', unmap row anyway, system may not recover object", index.row(), key);
 984         }
 985
 986         header->setKey(0);
 987         dirtyRow(accessor);
 988
 989         return true;
 990 }
 991
 992 /*
 993  * Get a mapped row
 994  * \param key is the 64 bits row key
 995  * Return a valid TRowIndex if success
 996  */
 997 RY_PDS::CObjectIndex    CTableBuffer::getMappedRow(uint64 key) const
 998 {
 999         if (!_Mapped)
1000         {
1001                 PDS_WARNING("mapRow(): table not mapped");
1002                 return RY_PDS::CObjectIndex::null();
1003         }
1004
1005         return _RowMapper.get(key);
1006 }
1007
1008
1009
1010 /*
1011  * Update common Timestamp
1012  */
1013 void    CTableBuffer::updateCommonStamp()
1014 {
1015         _CommonStamp = NLMISC::CTime::getSecondsSince1970();
1016 }
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026 /*
1027  * Setup debug delta file
1028  */
1029 bool    CTableBuffer::setupDebugDeltaFile(const std::string& filename, CDBDeltaFile& delta) const
1030 {
1031         delta.setup(filename, _InternalRowSize, CTimestamp(), CTimestamp());
1032
1033         if (!delta.preload())
1034         {
1035                 PDS_WARNING("setupDebugDeltaFile(): failed to preload file '%s'", filename.c_str());
1036                 return false;
1037         }
1038
1039         return true;
1040 }
1041
1042 /*
1043  * Get Delta file Row
1044  */
1045 uint8*  CTableBuffer::getDeltaRow(uint32& row, CDBDeltaFile& delta) const
1046 {
1047         static std::vector<uint8>       buffer;
1048
1049         buffer.resize(_InternalRowSize);
1050         uint8*  data = &(buffer[0]);
1051
1052         // read data from delta file
1053         if (!delta.read(row, data))
1054         {
1055                 PDS_WARNING("getDeltaRow(): failed to load next row from delta file");
1056                 return NULL;
1057         }
1058
1059         if (row == 0xffffffff)
1060                 return NULL;
1061
1062         return data + getHeaderSize();
1063 }