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