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