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