bundles/org.simantics.acorn/src/org/simantics/acorn/GraphClientImpl2.java

   1 /*******************************************************************************
   2  * Copyright (c) 2007, 2010 Association for Decentralized Information Management
   3  * in Industry THTH ry.
   4  * All rights reserved. This program and the accompanying materials
   5  * are made available under the terms of the Eclipse Public License v1.0
   6  * which accompanies this distribution, and is available at
   7  * http://www.eclipse.org/legal/epl-v10.html
   8  *
   9  * Contributors:
  10  *     VTT Technical Research Centre of Finland - initial API and implementation
  11  *******************************************************************************/
  12 package org.simantics.acorn;
  13
  14 import java.io.BufferedReader;
  15 import java.io.IOException;
  16 import java.nio.file.Files;
  17 import java.nio.file.Path;
  18 import java.util.ArrayList;
  19 import java.util.LinkedList;
  20 import java.util.concurrent.ExecutorService;
  21 import java.util.concurrent.Executors;
  22 import java.util.concurrent.Future;
  23 import java.util.concurrent.Semaphore;
  24 import java.util.concurrent.ThreadFactory;
  25 import java.util.concurrent.TimeUnit;
  26
  27 import org.simantics.acorn.MainProgram.MainProgramRunnable;
  28 import org.simantics.acorn.backup.AcornBackupProvider;
  29 import org.simantics.acorn.backup.AcornBackupProvider.AcornBackupRunnable;
  30 import org.simantics.acorn.exception.AcornAccessVerificationException;
  31 import org.simantics.acorn.exception.IllegalAcornStateException;
  32 import org.simantics.acorn.internal.ClusterChange;
  33 import org.simantics.acorn.internal.ClusterUpdateProcessorBase;
  34 import org.simantics.acorn.internal.UndoClusterUpdateProcessor;
  35 import org.simantics.acorn.lru.ClusterChangeSet.Entry;
  36 import org.simantics.acorn.lru.ClusterInfo;
  37 import org.simantics.acorn.lru.ClusterStreamChunk;
  38 import org.simantics.acorn.lru.ClusterUpdateOperation;
  39 import org.simantics.backup.BackupException;
  40 import org.simantics.db.ClusterCreator;
  41 import org.simantics.db.Database;
  42 import org.simantics.db.ServiceLocator;
  43 import org.simantics.db.exception.DatabaseException;
  44 import org.simantics.db.exception.SDBException;
  45 import org.simantics.db.server.ProCoreException;
  46 import org.simantics.db.service.ClusterSetsSupport;
  47 import org.simantics.db.service.ClusterUID;
  48 import org.simantics.db.service.LifecycleSupport;
  49 import org.simantics.utils.DataContainer;
  50 import org.simantics.utils.datastructures.Pair;
  51 import org.simantics.utils.logging.TimeLogger;
  52 import org.slf4j.Logger;
  53 import org.slf4j.LoggerFactory;
  54
  55 import gnu.trove.map.hash.TLongObjectHashMap;
  56
  57 public class GraphClientImpl2 implements Database.Session {
  58
  59     private static final Logger LOGGER = LoggerFactory.getLogger(GraphClientImpl2.class);
  60         public static final boolean DEBUG = false;
  61
  62         final ClusterManager clusters;
  63
  64         private TransactionManager transactionManager = new TransactionManager();
  65         private ExecutorService executor = Executors.newSingleThreadExecutor(new ClientThreadFactory("Core Main Program", false));
  66         private ExecutorService saver = Executors.newSingleThreadExecutor(new ClientThreadFactory("Core Snapshot Saver", true));
  67
  68         private Path dbFolder;
  69         private final Database database;
  70         private ServiceLocator locator;
  71         private MainProgram mainProgram;
  72
  73         private static class ClientThreadFactory implements ThreadFactory {
  74
  75                 final String name;
  76                 final boolean daemon;
  77
  78                 public ClientThreadFactory(String name, boolean daemon) {
  79                         this.name = name;
  80                         this.daemon = daemon;
  81                 }
  82
  83                 @Override
  84                 public Thread newThread(Runnable r) {
  85                         Thread thread = new Thread(r, name);
  86                         thread.setDaemon(daemon);
  87                         return thread;
  88                 }
  89         }
  90
  91         public GraphClientImpl2(Database database, Path dbFolder, ServiceLocator locator) throws IOException {
  92             this.database = database;
  93             this.dbFolder = dbFolder;
  94             this.locator = locator;
  95             this.clusters = new ClusterManager(dbFolder);
  96             load();
  97             ClusterSetsSupport cssi = locator.getService(ClusterSetsSupport.class);
  98             cssi.setReadDirectory(clusters.lastSessionDirectory);
  99             cssi.updateWriteDirectory(clusters.workingDirectory);
 100             mainProgram = new MainProgram(this, clusters);
 101             executor.execute(mainProgram);
 102         }
 103
 104         public Path getDbFolder() {
 105             return dbFolder;
 106         }
 107
 108         /*
 109          * This method schedules snapshotting.
 110          * No lock and thread restrictions.
 111          */
 112         void tryMakeSnapshot() throws IOException {
 113
 114                 if (isClosing || unexpectedClose)
 115                         return;
 116
 117                 saver.execute(new Runnable() {
 118
 119                         @Override
 120                         public void run() {
 121                                 Transaction tr = null;
 122                                 try {
 123                                         // First take a write transaction
 124                                         tr = askWriteTransaction(-1);
 125                                         // Then make sure that MainProgram is idling
 126                                         synchronizeWithIdleMainProgram(() -> makeSnapshot(false));
 127                                 } catch (IllegalAcornStateException | ProCoreException e) {
 128                                         LOGGER.error("Snapshotting failed", e);
 129                                         unexpectedClose = true;
 130                                 } catch (SDBException e) {
 131                                         LOGGER.error("Snapshotting failed", e);
 132                                         unexpectedClose = true;
 133                                 } finally {
 134                                         try {
 135                                                 if(tr != null)
 136                                                         endTransaction(tr.getTransactionId());
 137                                                 if (unexpectedClose) {
 138                                                         LifecycleSupport support = getServiceLocator().getService(LifecycleSupport.class);
 139                                                         try {
 140                                                                 support.close();
 141                                                         } catch (DatabaseException e1) {
 142                                                                 LOGGER.error("Failed to close database as a safety measure due to failed snapshotting", e1);
 143                                                         }
 144                                                 }
 145                                         } catch (ProCoreException e) {
 146                                             LOGGER.error("Failed to end snapshotting write transaction", e);
 147                                         }
 148                                 }
 149                         }
 150                 });
 151         }
 152
 153         private void makeSnapshot(boolean fullSave) throws IllegalAcornStateException {
 154                 clusters.makeSnapshot(locator, fullSave);
 155         }
 156
 157         @Override
 158         public <T> T clone(ClusterUID uid, ClusterCreator creator) throws DatabaseException {
 159             try {
 160             return clusters.clone(uid, creator);
 161         } catch (AcornAccessVerificationException | IllegalAcornStateException | IOException e) {
 162             unexpectedClose = true;
 163             throw new DatabaseException(e);
 164         }
 165         }
 166
 167         private void load() throws IOException {
 168                 clusters.load();
 169         }
 170
 171         @Override
 172         public Database getDatabase() {
 173                 return database;
 174         }
 175
 176         private boolean closed = false;
 177         private boolean isClosing = false;
 178         private boolean unexpectedClose = false;
 179
 180         @Override
 181         public void close() throws ProCoreException {
 182                 LOGGER.info("Closing " + this + " and mainProgram " + mainProgram);
 183                 if(!closed && !isClosing) {
 184                         isClosing = true;
 185                         try {
 186
 187                                 if (!unexpectedClose)
 188                                         synchronizeWithIdleMainProgram(() -> makeSnapshot(true));
 189
 190                                 mainProgram.close();
 191                                 clusters.shutdown();
 192                                 executor.shutdown();
 193                                 saver.shutdown();
 194
 195                                 boolean executorTerminated = executor.awaitTermination(500, TimeUnit.MILLISECONDS);
 196                                 boolean saverTerminated = saver.awaitTermination(500, TimeUnit.MILLISECONDS);
 197
 198                                 LOGGER.info("executorTerminated=" + executorTerminated + ", saverTerminated=" + saverTerminated);
 199
 200                                 try {
 201                                         clusters.mainState.save(dbFolder);
 202                                 } catch (IOException e) {
 203                                         LOGGER.error("Failed to save " + MainState.MAIN_STATE + " file in database folder " + dbFolder);
 204                                 }
 205
 206                                 mainProgram = null;
 207                                 executor = null;
 208                                 saver = null;
 209
 210                         } catch (IllegalAcornStateException | InterruptedException e) {
 211                                 throw new ProCoreException(e);
 212                         } catch (SDBException e1) {
 213                                 throw new ProCoreException(e1);
 214                         }
 215                         closed = true;
 216                 }
 217                 //impl.close();
 218         }
 219
 220         @Override
 221         public void open() throws ProCoreException {
 222                 throw new UnsupportedOperationException();
 223         }
 224
 225         @Override
 226         public boolean isClosed() throws ProCoreException {
 227                 return closed;
 228         }
 229
 230         @Override
 231         public void acceptCommit(long transactionId, long changeSetId, byte[] metadata) throws ProCoreException {
 232                 clusters.state.headChangeSetId++;
 233                 long committedChangeSetId = changeSetId + 1;
 234                 try {
 235                         clusters.commitChangeSet(committedChangeSetId, metadata);
 236                         clusters.state.transactionId = transactionId;
 237                         mainProgram.committed();
 238                         TimeLogger.log("Accepted commit");
 239                 } catch (IllegalAcornStateException e) {
 240                     throw new ProCoreException(e);
 241                 }
 242         }
 243
 244         @Override
 245         public long cancelCommit(long transactionId, long changeSetId, byte[] metadata, OnChangeSetUpdate onChangeSetUpdate) throws ProCoreException {
 246                 // Accept and finalize current transaction and then undo it
 247                 acceptCommit(transactionId, changeSetId, metadata);
 248
 249                 try {
 250                         undo(new long[] {changeSetId+1}, onChangeSetUpdate);
 251                         clusters.state.headChangeSetId++;
 252                         return clusters.state.headChangeSetId;
 253                 } catch (SDBException e) {
 254                         LOGGER.error("Failed to undo cancelled transaction", e);
 255                         throw new ProCoreException(e);
 256                 }
 257         }
 258
 259         @Override
 260         public Transaction askReadTransaction() throws ProCoreException {
 261                 return transactionManager.askReadTransaction();
 262         }
 263
 264         private enum TransactionState {
 265                 IDLE,WRITE,READ
 266         }
 267
 268         private class TransactionRequest {
 269                 public TransactionState state;
 270                 public Semaphore semaphore;
 271                 public TransactionRequest(TransactionState state, Semaphore semaphore) {
 272                         this.state = state;
 273                         this.semaphore = semaphore;
 274                 }
 275         }
 276
 277         private class TransactionManager {
 278
 279                 private TransactionState currentTransactionState = TransactionState.IDLE;
 280
 281                 private int reads = 0;
 282
 283                 private LinkedList<TransactionRequest> requests = new LinkedList<>();
 284
 285                 private TLongObjectHashMap<TransactionRequest> requestMap = new TLongObjectHashMap<>();
 286
 287                 private synchronized Transaction makeTransaction(TransactionRequest req) {
 288
 289                         final int csId = clusters.state.headChangeSetId;
 290                         final long trId = clusters.state.transactionId+1;
 291                         requestMap.put(trId, req);
 292                         return new Transaction() {
 293
 294                                 @Override
 295                                 public long getTransactionId() {
 296                                         return trId;
 297                                 }
 298
 299                                 @Override
 300                                 public long getHeadChangeSetId() {
 301                                         return csId;
 302                                 }
 303                         };
 304                 }
 305
 306                 /*
 307                  * This method cannot be synchronized since it waits and must support multiple entries
 308                  * by query thread(s) and internal transactions such as snapshot saver
 309                  */
 310                 private Transaction askReadTransaction() throws ProCoreException {
 311
 312                         Semaphore semaphore = new Semaphore(0);
 313
 314                         TransactionRequest req = queue(TransactionState.READ, semaphore);
 315
 316                         try {
 317                                 semaphore.acquire();
 318                         } catch (InterruptedException e) {
 319                                 throw new ProCoreException(e);
 320                         }
 321
 322                         return makeTransaction(req);
 323
 324                 }
 325
 326                 private synchronized void dispatch() {
 327                         TransactionRequest r = requests.removeFirst();
 328                         if(r.state == TransactionState.READ) reads++;
 329                         r.semaphore.release();
 330                 }
 331
 332                 private synchronized void processRequests() {
 333
 334                         while(true) {
 335
 336                                 if(requests.isEmpty()) return;
 337                                 TransactionRequest req = requests.peek();
 338
 339                                 if(currentTransactionState == TransactionState.IDLE) {
 340
 341                                         // Accept anything while IDLE
 342                                         currentTransactionState = req.state;
 343                                         dispatch();
 344
 345                                 } else if (currentTransactionState == TransactionState.READ) {
 346
 347                                         if(req.state == currentTransactionState) {
 348
 349                                                 // Allow other reads
 350                                                 dispatch();
 351
 352                                         } else {
 353
 354                                                 // Wait
 355                                                 return;
 356
 357                                         }
 358
 359                                 }  else if (currentTransactionState == TransactionState.WRITE) {
 360
 361                                         // Wait
 362                                         return;
 363
 364                                 }
 365
 366                         }
 367
 368                 }
 369
 370                 private synchronized TransactionRequest queue(TransactionState state, Semaphore semaphore) {
 371                         TransactionRequest req = new TransactionRequest(state, semaphore);
 372                         requests.addLast(req);
 373                         processRequests();
 374                         return req;
 375                 }
 376
 377                 /*
 378                  * This method cannot be synchronized since it waits and must support multiple entries
 379                  * by query thread(s) and internal transactions such as snapshot saver
 380                  */
 381                 private Transaction askWriteTransaction() throws IllegalAcornStateException {
 382
 383                         Semaphore semaphore = new Semaphore(0);
 384                         TransactionRequest req = queue(TransactionState.WRITE, semaphore);
 385
 386                         try {
 387                                 semaphore.acquire();
 388                         } catch (InterruptedException e) {
 389                                 throw new IllegalAcornStateException(e);
 390                         }
 391                         mainProgram.startTransaction(clusters.state.headChangeSetId+1);
 392                         return makeTransaction(req);
 393                 }
 394
 395                 private synchronized long endTransaction(long transactionId) throws ProCoreException {
 396
 397                         TransactionRequest req = requestMap.remove(transactionId);
 398                         if(req.state == TransactionState.WRITE) {
 399                                 currentTransactionState = TransactionState.IDLE;
 400                                 processRequests();
 401                         } else {
 402                                 reads--;
 403                                 if(reads == 0) {
 404                                         currentTransactionState = TransactionState.IDLE;
 405                                         processRequests();
 406                                 }
 407                         }
 408                         return clusters.state.transactionId;
 409                 }
 410
 411         }
 412
 413         @Override
 414         public Transaction askWriteTransaction(final long transactionId) throws ProCoreException {
 415                 try {
 416                     if (isClosing || unexpectedClose || closed) {
 417                         throw new ProCoreException("GraphClientImpl2 is already closing so no more write transactions allowed!");
 418                     }
 419             return transactionManager.askWriteTransaction();
 420         } catch (IllegalAcornStateException e) {
 421             throw new ProCoreException(e);
 422         }
 423         }
 424
 425         @Override
 426         public long endTransaction(long transactionId) throws ProCoreException {
 427                 return transactionManager.endTransaction(transactionId);
 428         }
 429
 430         @Override
 431         public String execute(String command) throws ProCoreException {
 432                 // This is called only by WriteGraphImpl.commitAccessorChanges
 433                 // We can ignore this in Acorn
 434                 return "";
 435         }
 436
 437         @Override
 438         public byte[] getChangeSetMetadata(long changeSetId) throws ProCoreException {
 439                 try {
 440             return clusters.getMetadata(changeSetId);
 441         } catch (AcornAccessVerificationException | IllegalAcornStateException e) {
 442             throw new ProCoreException(e);
 443         }
 444         }
 445
 446         @Override
 447         public ChangeSetData getChangeSetData(long minChangeSetId,
 448                         long maxChangeSetId, OnChangeSetUpdate onChangeSetupate)
 449                         throws ProCoreException {
 450
 451                 new Exception("GetChangeSetDataFunction " + minChangeSetId + " " + maxChangeSetId).printStackTrace();;
 452                 return null;
 453
 454         }
 455
 456         @Override
 457         public ChangeSetIds getChangeSetIds() throws ProCoreException {
 458                 throw new UnsupportedOperationException();
 459         }
 460
 461         @Override
 462         public Cluster getCluster(byte[] clusterId) throws ProCoreException {
 463                 throw new UnsupportedOperationException();
 464         }
 465
 466         @Override
 467         public ClusterChanges getClusterChanges(long changeSetId, byte[] clusterId)
 468                         throws ProCoreException {
 469                 throw new UnsupportedOperationException();
 470         }
 471
 472         @Override
 473         public ClusterIds getClusterIds() throws ProCoreException {
 474                 try {
 475             return clusters.getClusterIds();
 476         } catch (IllegalAcornStateException e) {
 477             throw new ProCoreException(e);
 478         }
 479         }
 480
 481         @Override
 482         public Information getInformation() throws ProCoreException {
 483                 return new Information() {
 484
 485                         @Override
 486                         public String getServerId() {
 487                                 return "server";
 488                         }
 489
 490                         @Override
 491                         public String getProtocolId() {
 492                                 return "";
 493                         }
 494
 495                         @Override
 496                         public String getDatabaseId() {
 497                                 return "database";
 498                         }
 499
 500                         @Override
 501                         public long getFirstChangeSetId() {
 502                                 return 0;
 503                         }
 504
 505                 };
 506         }
 507
 508         @Override
 509         public Refresh getRefresh(long changeSetId) throws ProCoreException {
 510
 511                 final ClusterIds ids = getClusterIds();
 512
 513                 return new Refresh() {
 514
 515                         @Override
 516                         public long getHeadChangeSetId() {
 517                                 return clusters.state.headChangeSetId;
 518                         }
 519
 520                         @Override
 521                         public long[] getFirst() {
 522                                 return ids.getFirst();
 523                         }
 524
 525                         @Override
 526                         public long[] getSecond() {
 527                                 return ids.getSecond();
 528                         }
 529
 530                 };
 531
 532         }
 533
 534 //      public byte[] getResourceFile(final byte[] clusterUID, final int resourceIndex) throws ProCoreException, AcornAccessVerificationException, IllegalAcornStateException {
 535 //              return clusters.getResourceFile(clusterUID, resourceIndex);
 536 //      }
 537
 538         @Override
 539         public ResourceSegment getResourceSegment(final byte[] clusterUID, final int resourceIndex, final long segmentOffset, short segmentSize) throws ProCoreException {
 540                 try {
 541             return clusters.getResourceSegment(clusterUID, resourceIndex, segmentOffset, segmentSize);
 542         } catch (AcornAccessVerificationException | IllegalAcornStateException e) {
 543             throw new ProCoreException(e);
 544         }
 545         }
 546
 547         @Override
 548         public long reserveIds(int count) throws ProCoreException {
 549                 return clusters.state.reservedIds++;
 550         }
 551
 552         @Override
 553         public void updateCluster(byte[] operations) throws ProCoreException {
 554             ClusterInfo info = null;
 555             try {
 556                 ClusterUpdateOperation operation = new ClusterUpdateOperation(clusters, operations);
 557                 info = clusters.clusterLRU.getOrCreate(operation.uid, true);
 558                 if(info == null)
 559                     throw new IllegalAcornStateException("info == null for operation " + operation);
 560                 info.acquireMutex();
 561                         info.scheduleUpdate();
 562                         mainProgram.schedule(operation);
 563                 } catch (IllegalAcornStateException | AcornAccessVerificationException e) {
 564             throw new ProCoreException(e);
 565         } finally {
 566             if (info != null)
 567                 info.releaseMutex();
 568                 }
 569         }
 570
 571         private UndoClusterUpdateProcessor getUndoCSS(String ccsId) throws DatabaseException, AcornAccessVerificationException, IllegalAcornStateException {
 572
 573                 String[] ss = ccsId.split("\\.");
 574                 String chunkKey = ss[0];
 575                 int chunkOffset = Integer.parseInt(ss[1]);
 576                 ClusterStreamChunk chunk = clusters.streamLRU.getWithoutMutex(chunkKey);
 577                 if(chunk == null) throw new IllegalAcornStateException("Cluster Stream Chunk " + chunkKey + " was not found.");
 578                 chunk.acquireMutex();
 579                 try {
 580                         return chunk.getUndoProcessor(clusters, chunkOffset, ccsId);
 581                 } catch (DatabaseException e) {
 582                     throw e;
 583                 } catch (Throwable t) {
 584                         throw new IllegalStateException(t);
 585                 } finally {
 586                         chunk.releaseMutex();
 587                 }
 588         }
 589
 590         private void performUndo(String ccsId, ArrayList<Pair<ClusterUID, byte[]>> clusterChanges, UndoClusterSupport support) throws ProCoreException, DatabaseException, IllegalAcornStateException, AcornAccessVerificationException {
 591                 UndoClusterUpdateProcessor proc = getUndoCSS(ccsId);
 592
 593                 int clusterKey = clusters.getClusterKeyByClusterUIDOrMakeWithoutMutex(proc.getClusterUID());
 594
 595                 clusters.clusterLRU.acquireMutex();
 596                 try {
 597
 598                         ClusterChange cs = new ClusterChange(clusterChanges, proc.getClusterUID());
 599                         for(int i=0;i<proc.entries.size();i++) {
 600
 601                                 Entry e = proc.entries.get(proc.entries.size() - 1 - i);
 602                                 e.process(clusters, cs, clusterKey);
 603                         }
 604                         cs.flush();
 605
 606                 } finally {
 607                         clusters.clusterLRU.releaseMutex();
 608                 }
 609         }
 610
 611         private void synchronizeWithIdleMainProgram(MainProgramRunnable runnable) throws SDBException {
 612
 613                 Exception[] exception = { null };
 614                 Semaphore s = new Semaphore(0);
 615
 616                 mainProgram.runIdle(new MainProgramRunnable() {
 617
 618                         @Override
 619                         public void success() {
 620                                 try {
 621                                     runnable.success();
 622                                 } finally {
 623                                     s.release();
 624                                 }
 625                         }
 626
 627                         @Override
 628                         public void error(Exception e) {
 629                                 exception[0] = e;
 630                                 try {
 631                                     runnable.error(e);
 632                                 } finally {
 633                                     s.release();
 634                                 }
 635                         }
 636
 637                         @Override
 638                         public void run() throws Exception {
 639                                 runnable.run();
 640                         }
 641
 642                 });
 643
 644                 try {
 645                         s.acquire();
 646                 } catch (InterruptedException e) {
 647                         throw new IllegalAcornStateException("Unhandled interruption.", e);
 648                 }
 649
 650                 Exception e = exception[0];
 651                 if(e != null) {
 652                         if(e instanceof SDBException) throw (SDBException)e;
 653                         else if(e != null) throw new IllegalAcornStateException(e);
 654                 }
 655
 656         }
 657
 658         @Override
 659         public boolean undo(long[] changeSetIds, OnChangeSetUpdate onChangeSetUpdate) throws SDBException {
 660
 661                 synchronizeWithIdleMainProgram(new MainProgramRunnable() {
 662
 663                         @Override
 664                         public void run() throws Exception {
 665
 666                                 try {
 667
 668                                 final ArrayList<Pair<ClusterUID, byte[]>> clusterChanges = new ArrayList<Pair<ClusterUID, byte[]>>();
 669
 670                                 UndoClusterSupport support = new UndoClusterSupport(clusters);
 671
 672                                 final int changeSetId = clusters.state.headChangeSetId;
 673
 674                                 if(ClusterUpdateProcessorBase.DEBUG)
 675                                         LOGGER.info(" === BEGIN UNDO ===");
 676
 677                                 for(int i=0;i<changeSetIds.length;i++) {
 678                                         final long id = changeSetIds[changeSetIds.length-1-i];
 679                                         ArrayList<String> ccss = clusters.getChanges(id);
 680
 681                                         for(int j=0;j<ccss.size();j++) {
 682                                                 String ccsid = ccss.get(ccss.size()-j-1);
 683                                                 try {
 684                                                         if(ClusterUpdateProcessorBase.DEBUG)
 685                                                                 LOGGER.info("performUndo " + ccsid);
 686                                                         performUndo(ccsid, clusterChanges, support);
 687                                                 } catch (DatabaseException e) {
 688                                                         e.printStackTrace();
 689                                                 }
 690                                         }
 691                                 }
 692
 693                                 if(ClusterUpdateProcessorBase.DEBUG)
 694                                         LOGGER.info(" === END UNDO ===");
 695
 696                                 for(int i=0;i<clusterChanges.size();i++) {
 697
 698                                         final int changeSetIndex = i;
 699
 700                                         final Pair<ClusterUID, byte[]> pair = clusterChanges.get(i);
 701
 702                                         final ClusterUID cuid = pair.first;
 703                                         final byte[] data = pair.second;
 704
 705                                         onChangeSetUpdate.onChangeSetUpdate(new ChangeSetUpdate() {
 706
 707                                                 @Override
 708                                                 public long getChangeSetId() {
 709                                                         return changeSetId;
 710                                                 }
 711
 712                                                 @Override
 713                                                 public int getChangeSetIndex() {
 714                                                         return 0;
 715                                                 }
 716
 717                                                 @Override
 718                                                 public int getNumberOfClusterChangeSets() {
 719                                                         return clusterChanges.size();
 720                                                 }
 721
 722                                                 @Override
 723                                                 public int getIndexOfClusterChangeSet() {
 724                                                         return changeSetIndex;
 725                                                 }
 726
 727                                                 @Override
 728                                                 public byte[] getClusterId() {
 729                                                         return cuid.asBytes();
 730                                                 }
 731
 732                                                 @Override
 733                                                 public boolean getNewCluster() {
 734                                                         return false;
 735                                                 }
 736
 737                                                 @Override
 738                                                 public byte[] getData() {
 739                                                         return data;
 740                                                 }
 741
 742                                         });
 743                                 }
 744                         } catch (AcornAccessVerificationException | IllegalAcornStateException e1) {
 745                             throw new ProCoreException(e1);
 746                         }
 747
 748                         }
 749
 750                 });
 751
 752                 return false;
 753
 754         }
 755
 756         ServiceLocator getServiceLocator() {
 757             return locator;
 758         }
 759
 760     @Override
 761     public boolean refreshEnabled() {
 762         return false;
 763     }
 764
 765     @Override
 766     public boolean rolledback() {
 767         return clusters.rolledback();
 768     }
 769
 770     private void purge() throws IllegalAcornStateException {
 771         clusters.purge(locator);
 772     }
 773
 774     public void purgeDatabase() {
 775
 776             if (isClosing || unexpectedClose)
 777                 return;
 778
 779                 saver.execute(new Runnable() {
 780
 781                         @Override
 782                         public void run() {
 783                                 Transaction tr = null;
 784                                 try {
 785                                         // First take a write transaction
 786                                         tr = askWriteTransaction(-1);
 787                                         // Then make sure that MainProgram is idling
 788                                         synchronizeWithIdleMainProgram(() -> purge());
 789                                 } catch (IllegalAcornStateException | ProCoreException e) {
 790                                         LOGGER.error("Purge failed", e);
 791                                         unexpectedClose = true;
 792                                 } catch (SDBException e) {
 793                                         LOGGER.error("Purge failed", e);
 794                                         unexpectedClose = true;
 795                                 } finally {
 796                                         try {
 797                                                 if(tr != null)
 798                                                         endTransaction(tr.getTransactionId());
 799                                                 if (unexpectedClose) {
 800                                                         LifecycleSupport support = getServiceLocator().getService(LifecycleSupport.class);
 801                                                         try {
 802                                                                 support.close();
 803                                                         } catch (DatabaseException e1) {
 804                                                                 LOGGER.error("Failed to close database as a safety measure due to failed purge", e1);
 805                                                         }
 806                                                 }
 807                                         } catch (ProCoreException e) {
 808                                             LOGGER.error("Failed to end purge write transaction", e);
 809                                         }
 810                                 }
 811                         }
 812                 });
 813
 814     }
 815
 816     public long getTailChangeSetId() {
 817         return clusters.getTailChangeSetId();
 818     }
 819
 820         public Future<BackupException> getBackupRunnable(Semaphore lock, Path targetPath, int revision) throws IllegalAcornStateException, IOException {
 821
 822                 makeSnapshot(true);
 823
 824                 Path dbDir = getDbFolder();
 825                 int newestFolder = clusters.mainState.headDir - 1;
 826                 int latestFolder = -2;
 827                 Path AcornMetadataFile = AcornBackupProvider.getAcornMetadataFile(dbDir);
 828                 if (Files.exists(AcornMetadataFile)) {
 829                         try (BufferedReader br = Files.newBufferedReader(AcornMetadataFile)) {
 830                                 latestFolder = Integer.parseInt( br.readLine() );
 831                         }
 832                 }
 833
 834                 AcornBackupRunnable r = new AcornBackupRunnable(
 835                                 lock, targetPath, revision, dbDir, latestFolder, newestFolder);
 836                 new Thread(r, "Acorn backup thread").start();
 837                 return r;
 838
 839         }
 840
 841 }
 842