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

   1 package org.simantics.acorn;
   2
   3 import java.io.IOException;
   4 import java.math.BigInteger;
   5 import java.nio.file.DirectoryStream;
   6 import java.nio.file.Files;
   7 import java.nio.file.Path;
   8 import java.util.ArrayList;
   9 import java.util.Collection;
  10 import java.util.HashMap;
  11 import java.util.Map;
  12
  13 import org.simantics.acorn.cluster.ClusterImpl;
  14 import org.simantics.acorn.internal.ClusterSupport2;
  15 import org.simantics.acorn.lru.ChangeSetInfo;
  16 import org.simantics.acorn.lru.ClusterInfo;
  17 import org.simantics.acorn.lru.ClusterLRU;
  18 import org.simantics.acorn.lru.ClusterStreamChunk;
  19 import org.simantics.acorn.lru.FileInfo;
  20 import org.simantics.acorn.lru.LRU;
  21 import org.simantics.db.ClusterCreator;
  22 import org.simantics.db.ServiceLocator;
  23 import org.simantics.db.Database.Session.ClusterIds;
  24 import org.simantics.db.Database.Session.ResourceSegment;
  25 import org.simantics.db.exception.DatabaseException;
  26 import org.simantics.db.impl.ClusterBase;
  27 import org.simantics.db.impl.ClusterI;
  28 import org.simantics.db.impl.ClusterSupport;
  29 import org.simantics.db.procore.cluster.ClusterTraits;
  30 import org.simantics.db.server.ProCoreException;
  31 import org.simantics.db.service.ClusterSetsSupport;
  32 import org.simantics.db.service.ClusterUID;
  33 import org.simantics.utils.threads.logger.ITask;
  34 import org.simantics.utils.threads.logger.ThreadLogger;
  35
  36 public class ClusterManager {
  37
  38         private ArrayList<String> currentChanges = new ArrayList<String>();
  39
  40         public final Path dbFolder;
  41         public Path lastSessionDirectory;
  42         public Path workingDirectory;
  43
  44         public LRU<String, ClusterStreamChunk> streamLRU;
  45         public LRU<Long, ChangeSetInfo> csLRU;
  46         public ClusterLRU clusterLRU;
  47         public LRU<String, FileInfo> fileLRU;
  48
  49         public MainState mainState;
  50         public HeadState state;
  51
  52         private long lastSnapshot = System.nanoTime();
  53
  54         final public ClusterSupport2 support = new ClusterSupport2(this);
  55
  56         /*
  57          * Public interface
  58          *
  59          */
  60
  61         public ClusterManager(Path dbFolder) {
  62                 this.dbFolder = dbFolder;
  63         }
  64
  65         public ArrayList<String> getChanges(long changeSetId) {
  66                 ChangeSetInfo info = csLRU.getWithoutMutex(changeSetId);
  67                 info.acquireMutex();
  68                 try {
  69                         info.makeResident();
  70                         return info.getCSSIds();
  71                 } catch (Throwable t) {
  72                         throw new IllegalStateException(t);
  73                 } finally {
  74                         info.releaseMutex();
  75                 }
  76         }
  77
  78         public ClusterBase getClusterByClusterKey(int clusterKey) throws DatabaseException {
  79                 return clusterLRU.getClusterByClusterKey(clusterKey);
  80         }
  81
  82         public ClusterBase getClusterByClusterUIDOrMake(ClusterUID clusterUID) throws DatabaseException {
  83                 return clusterLRU.getClusterByClusterUIDOrMake(clusterUID);
  84         }
  85
  86         public ClusterImpl getClusterByClusterUIDOrMakeProxy(ClusterUID clusterUID) throws DatabaseException {
  87                 return clusterLRU.getClusterByClusterUIDOrMakeProxy(clusterUID);
  88         }
  89
  90         public int getClusterKeyByClusterUIDOrMake(ClusterUID clusterUID) {
  91                 return clusterLRU.getClusterKeyByClusterUIDOrMake(clusterUID);
  92         }
  93
  94         public int getClusterKeyByClusterUIDOrMakeWithoutMutex(ClusterUID clusterUID) {
  95                 return clusterLRU.getClusterKeyByClusterUIDOrMakeWithoutMutex(clusterUID);
  96         }
  97
  98         public int getClusterKeyByUID(long id1, long id2) throws DatabaseException {
  99                 return clusterLRU.getClusterKeyByUIDWithoutMutex(id1, id2);
 100         }
 101
 102         public <T extends ClusterI> T getClusterProxyByResourceKey(int resourceKey) throws DatabaseException {
 103                 return clusterLRU.getClusterProxyByResourceKey(resourceKey);
 104         }
 105
 106         public ClusterUID getClusterUIDByResourceKey(int resourceKey) throws DatabaseException {
 107                 return clusterLRU.getClusterUIDByResourceKey(resourceKey);
 108         }
 109
 110         public ClusterUID getClusterUIDByResourceKeyWithoutMutex(int resourceKey) throws DatabaseException {
 111                 return clusterLRU.getClusterUIDByResourceKeyWithoutMutex(resourceKey);
 112         }
 113
 114         /*
 115          * Private implementation
 116          *
 117          */
 118
 119         private static long countFiles(Path directory) throws IOException {
 120                 try (DirectoryStream<Path> ds = Files.newDirectoryStream(directory)) {
 121                         int count = 0;
 122                         for (@SuppressWarnings("unused") Path p : ds)
 123                                 ++count;
 124                         return count;
 125                 }
 126         }
 127
 128         public synchronized boolean makeSnapshot(ServiceLocator locator, boolean force) throws IOException {
 129
 130                 // Maximum autosave frequency is per 60s
 131                 if(!force && System.nanoTime() - lastSnapshot < 10*1000000000L) {
 132 //                  System.err.println("lastSnapshot too early");
 133                     return false;
 134                 }
 135
 136                 // Cluster files are always there
 137                 // Nothing has been written => no need to do anything
 138                 long amountOfFiles = countFiles(workingDirectory);
 139                 if(!force && amountOfFiles < 3) {
 140 //                  System.err.println("amountOfFiles < 3");
 141                     return false;
 142                 }
 143
 144                 System.err.println("makeSnapshot");
 145
 146                 // Schedule writing of all data to disk
 147                 refreshHeadState();
 148
 149                 // Wait for all files to be written
 150                 clusterLRU.shutdown();
 151                 fileLRU.shutdown();
 152                 streamLRU.shutdown();
 153                 csLRU.shutdown();
 154
 155                 persistHeadState();
 156
 157                 mainState.save(dbFolder);
 158
 159                 ClusterSetsSupport cssi = locator.getService(ClusterSetsSupport.class);
 160                 cssi.save();
 161
 162                 amountOfFiles = countFiles(workingDirectory);
 163
 164                 System.err.println(" -finished: amount of files is " + amountOfFiles);
 165
 166                 workingDirectory = dbFolder.resolve(Integer.toString(mainState.headDir));
 167                 if (!Files.exists(workingDirectory)) {
 168                     Files.createDirectories(workingDirectory);
 169                 }
 170
 171                 cssi.updateReadAndWriteDirectories(lastSessionDirectory, workingDirectory);
 172
 173                 clusterLRU.setWriteDir(workingDirectory);
 174                 fileLRU.setWriteDir(workingDirectory);
 175                 streamLRU.setWriteDir(workingDirectory);
 176                 csLRU.setWriteDir(workingDirectory);
 177
 178                 clusterLRU.resume();
 179                 fileLRU.resume();
 180                 streamLRU.resume();
 181                 csLRU.resume();
 182
 183                 lastSnapshot = System.nanoTime();
 184
 185                 return true;
 186
 187         }
 188
 189         public void refreshHeadState() throws IOException {
 190
 191                 state.clusters.clear();
 192                 state.files.clear();
 193                 state.stream.clear();
 194                 state.cs.clear();
 195
 196                 clusterLRU.persist(state.clusters);
 197                 fileLRU.persist(state.files);
 198                 streamLRU.persist(state.stream);
 199                 csLRU.persist(state.cs);
 200
 201         }
 202
 203         public void persistHeadState() throws IOException {
 204
 205                 // Sync current working directory
 206                 Files.walk(workingDirectory, 1).filter(Files::isRegularFile).forEach(FileIO::uncheckedSyncPath);
 207                 state.save(workingDirectory);
 208                 mainState.headDir++;
 209         }
 210
 211
 212 //      public void save() throws IOException {
 213 //
 214 //              refreshHeadState();
 215 //
 216 //              clusterLRU.shutdown();
 217 //              fileLRU.shutdown();
 218 //              streamLRU.shutdown();
 219 //              csLRU.shutdown();
 220 //
 221 //              persistHeadState();
 222 //
 223 //              mainState.save(getBaseDirectory());
 224
 225 //              try {
 226 //                      ThreadLogVisualizer visualizer = new ThreadLogVisualizer();
 227 //                      visualizer.read(new DataInputStream(new FileInputStream(
 228 //                                      ThreadLogger.LOG_FILE)));
 229 //                      visualizer.visualize3(new PrintStream(ThreadLogger.LOG_FILE
 230 //                                      + ".svg"));
 231 //              } catch (FileNotFoundException e) {
 232 //                      // TODO Auto-generated catch block
 233 //                      e.printStackTrace();
 234 //              }
 235
 236                 // System.err.println("-- load statistics --");
 237                 // for(Pair<ClusterUID, Integer> entry :
 238                 // CollectionUtils.valueSortedEntries(histogram)) {
 239                 // System.err.println(" " + entry.second + " " + entry.first);
 240                 // }
 241
 242 //      }
 243
 244         private void acquireAll() {
 245                 clusterLRU.acquireMutex();
 246                 fileLRU.acquireMutex();
 247                 streamLRU.acquireMutex();
 248                 csLRU.acquireMutex();
 249         }
 250
 251         private void releaseAll() {
 252                 csLRU.releaseMutex();
 253                 streamLRU.releaseMutex();
 254                 fileLRU.releaseMutex();
 255                 clusterLRU.releaseMutex();
 256         }
 257
 258         public void load() throws IOException {
 259
 260                 // Main state
 261                 mainState = MainState.load(dbFolder);
 262
 263                 lastSessionDirectory = dbFolder.resolve(Integer.toString(mainState.headDir - 1));
 264
 265                 // Head State
 266                 try {
 267             state = HeadState.load(lastSessionDirectory);
 268         } catch (InvalidHeadStateException e) {
 269             // For backwards compatibility only!
 270             Throwable cause = e.getCause();
 271             if (cause instanceof Throwable) {
 272                 try {
 273                     org.simantics.db.javacore.HeadState oldState = org.simantics.db.javacore.HeadState.load(lastSessionDirectory);
 274
 275                     HeadState newState = new HeadState();
 276                     newState.clusters = oldState.clusters;
 277                     newState.cs = oldState.cs;
 278                     newState.files = oldState.files;
 279                     newState.stream = oldState.stream;
 280                     newState.headChangeSetId = oldState.headChangeSetId;
 281                     newState.reservedIds = oldState.reservedIds;
 282                     newState.transactionId = oldState.transactionId;
 283                     state = newState;
 284                 } catch (InvalidHeadStateException e1) {
 285                     throw new IOException("Could not load HeadState due to corruption", e1);
 286                 }
 287             } else {
 288                 // This should never happen as MainState.load() checks the integrity
 289                 // of head.state files and rolls back in cases of corruption until a
 290                 // consistent state is found (could be case 0 - initial db state)
 291                 // IF this does happen something is completely wrong
 292                 throw new IOException("Could not load HeadState due to corruption", e);
 293             }
 294         }
 295
 296                 workingDirectory = dbFolder.resolve(Integer.toString(mainState.headDir));
 297                 Files.createDirectories(workingDirectory);
 298
 299                 csLRU = new LRU<Long, ChangeSetInfo>("Change Set", workingDirectory);
 300                 streamLRU = new LRU<String, ClusterStreamChunk>("Cluster Stream", workingDirectory);
 301                 clusterLRU = new ClusterLRU(this, "Cluster", workingDirectory);
 302                 fileLRU = new LRU<String, FileInfo>("External Value", workingDirectory);
 303
 304                 acquireAll();
 305
 306                 // Clusters
 307                 for (String clusterKey : state.clusters) {
 308                         String[] parts1 = clusterKey.split("#");
 309                         String[] parts = parts1[0].split("\\.");
 310                         long first = new BigInteger(parts[0], 16).longValue();
 311                         long second = new BigInteger(parts[1], 16).longValue();
 312                         ClusterUID uuid = ClusterUID.make(first, second);
 313                         Path readDir = dbFolder.resolve(parts1[1]);
 314                         int offset = Integer.parseInt(parts1[2]);
 315                         int length = Integer.parseInt(parts1[3]);
 316                         clusterLRU.map(new ClusterInfo(this, clusterLRU, readDir, uuid, offset, length));
 317                 }
 318                 // Files
 319                 for (String fileKey : state.files) {
 320 //                      System.err.println("loadFile: " + fileKey);
 321                         String[] parts = fileKey.split("#");
 322                         Path readDir = dbFolder.resolve(parts[1]);
 323                         int offset = Integer.parseInt(parts[2]);
 324                         int length = Integer.parseInt(parts[3]);
 325                         FileInfo info = new FileInfo(fileLRU, readDir, parts[0], offset, length);
 326                         fileLRU.map(info);
 327                 }
 328                 // Update chunks
 329                 for (String fileKey : state.stream) {
 330 //                      System.err.println("loadStream: " + fileKey);
 331                         String[] parts = fileKey.split("#");
 332                         Path readDir = dbFolder.resolve(parts[1]);
 333                         int offset = Integer.parseInt(parts[2]);
 334                         int length = Integer.parseInt(parts[3]);
 335                         ClusterStreamChunk info = new ClusterStreamChunk(this,
 336                                         streamLRU, readDir, parts[0], offset, length);
 337                         streamLRU.map(info);
 338                 }
 339                 // Change sets
 340                 for (String fileKey : state.cs) {
 341                         String[] parts = fileKey.split("#");
 342                         Path readDir = dbFolder.resolve(parts[1]);
 343                         Long revisionId = Long.parseLong(parts[0]);
 344                         int offset = Integer.parseInt(parts[2]);
 345                         int length = Integer.parseInt(parts[3]);
 346                         ChangeSetInfo info = new ChangeSetInfo(csLRU, readDir, revisionId, offset, length);
 347                         csLRU.map(info);
 348                 }
 349
 350                 releaseAll();
 351
 352         }
 353
 354         public <T> T clone(ClusterUID uid, ClusterCreator creator)
 355                         throws DatabaseException {
 356
 357                 clusterLRU.ensureUpdates(uid);
 358
 359                 ClusterInfo info = clusterLRU.getWithoutMutex(uid);
 360                 return info.clone(uid, creator);
 361
 362         }
 363
 364         //private int loadCounter = 0;
 365
 366         public static void startLog(String msg) {
 367                 tasks.put(msg, ThreadLogger.getInstance().begin(msg));
 368         }
 369
 370         public static void endLog(String msg) {
 371                 ITask task = tasks.get(msg);
 372                 if (task != null)
 373                         task.finish();
 374         }
 375
 376         static Map<String, ITask> tasks = new HashMap<String, ITask>();
 377
 378         public void update(ClusterUID uid, ClusterImpl clu) {
 379
 380                 ClusterInfo info = clusterLRU.getWithoutMutex(uid);
 381                 info.acquireMutex();
 382                 try {
 383                         info.update(clu);
 384                 } catch (Throwable t) {
 385                         throw new IllegalStateException(t);
 386                 } finally {
 387                         info.releaseMutex();
 388                 }
 389
 390         }
 391
 392         public long getClusterIdOrCreate(ClusterUID clusterUID) {
 393                 return 1;
 394         }
 395
 396         public int getResourceKey(ClusterUID uid, int index) {
 397                 return clusterLRU.getResourceKey(uid, index);
 398         }
 399
 400         public int getResourceKeyWitoutMutex(ClusterUID uid, int index) {
 401                 return clusterLRU.getResourceKeyWithoutMutex(uid, index);
 402         }
 403
 404         public ClusterIds getClusterIds() throws ProCoreException {
 405
 406                 clusterLRU.acquireMutex();
 407
 408                 try {
 409
 410                         Collection<ClusterInfo> infos = clusterLRU.values();
 411                         final int status = infos.size();
 412                         final long[] firsts = new long[status];
 413                         final long[] seconds = new long[status];
 414
 415                         int index = 0;
 416                         for (ClusterInfo info : infos) {
 417                                 firsts[index] = 0;
 418                                 seconds[index] = info.getKey().second;
 419                                 index++;
 420                         }
 421
 422                         return new ClusterIds() {
 423
 424                                 @Override
 425                                 public int getStatus() {
 426                                         return status;
 427                                 }
 428
 429                                 @Override
 430                                 public long[] getFirst() {
 431                                         return firsts;
 432                                 }
 433
 434                                 @Override
 435                                 public long[] getSecond() {
 436                                         return seconds;
 437                                 }
 438
 439                         };
 440
 441                 } catch (Throwable t) {
 442                         throw new IllegalStateException(t);
 443                 } finally {
 444                         clusterLRU.releaseMutex();
 445                 }
 446
 447         }
 448
 449         public void addIntoCurrentChangeSet(String ccs) {
 450
 451                 csLRU.acquireMutex();
 452
 453                 try {
 454
 455                         currentChanges.add(ccs);
 456
 457                 } catch (Throwable t) {
 458                         throw new IllegalStateException(t);
 459                 } finally {
 460
 461                         csLRU.releaseMutex();
 462
 463                 }
 464
 465         }
 466
 467         public void commitChangeSet(long changeSetId, byte[] data) {
 468                 csLRU.acquireMutex();
 469                 try {
 470                         ArrayList<String> csids = new ArrayList<String>(currentChanges);
 471                         currentChanges = new ArrayList<String>();
 472                         new ChangeSetInfo(csLRU, changeSetId, data, csids);
 473                 } catch (Throwable t) {
 474                         throw new IllegalStateException(t);
 475                 } finally {
 476                         csLRU.releaseMutex();
 477                 }
 478         }
 479
 480         public byte[] getMetadata(long changeSetId) {
 481
 482                 ChangeSetInfo info = csLRU.getWithoutMutex(changeSetId);
 483                 if (info == null) return null;
 484                 info.acquireMutex();
 485                 try {
 486                         return info.getMetadataBytes();
 487                 } catch (Throwable t) {
 488                         throw new IllegalStateException(t);
 489                 } finally {
 490                         info.releaseMutex();
 491                 }
 492
 493         }
 494
 495         public byte[] getResourceFile(final byte[] clusterUID,
 496                         final int resourceIndex) throws ProCoreException {
 497
 498                 ClusterUID uid = ClusterUID.make(clusterUID, 0);
 499                 String key = uid.toString() + "_" + resourceIndex;
 500                 FileInfo info = fileLRU.getWithoutMutex(key);
 501                 if(info == null) return null;
 502                 info.acquireMutex();
 503                 try {
 504                         return info.getResourceFile();
 505                 } catch (Throwable t) {
 506                         throw new IllegalStateException(t);
 507                 } finally {
 508                         info.releaseMutex();
 509                 }
 510
 511         }
 512
 513         public ResourceSegment getResourceSegment(final byte[] clusterUID,
 514                         final int resourceIndex, final long segmentOffset, short segmentSize)
 515                         throws ProCoreException {
 516
 517                 ClusterUID uid = ClusterUID.make(clusterUID, 0);
 518
 519                 String key = uid.toString() + "_" + resourceIndex;
 520                 FileInfo info = fileLRU.getWithoutMutex(key);
 521                 if(info == null) return null;
 522                 info.acquireMutex();
 523                 try {
 524                         return info.getResourceSegment(clusterUID, resourceIndex, segmentOffset, segmentSize);
 525                 } catch (Throwable t) {
 526                         throw new IllegalStateException(t);
 527                 } finally {
 528                         info.releaseMutex();
 529                 }
 530
 531         }
 532
 533         public void modiFileEx(ClusterUID uid, int resourceKey, long offset,
 534                         long size, byte[] bytes, long pos, ClusterSupport support) {
 535
 536                 try {
 537
 538                         String key = uid.toString()
 539                                         + "_"
 540                                         + ClusterTraits
 541                                                         .getResourceIndexFromResourceKey(resourceKey);
 542
 543                         FileInfo info = null;
 544
 545                         fileLRU.acquireMutex();
 546
 547                         try {
 548
 549                                 info = fileLRU.get(key);
 550                                 if (info == null)
 551                                         info = new FileInfo(fileLRU, key, (int) (offset + size));
 552
 553
 554                         } catch (Throwable t) {
 555                                 throw new IllegalStateException(t);
 556                         } finally {
 557
 558                                 fileLRU.releaseMutex();
 559
 560                         }
 561
 562                         info.acquireMutex();
 563                         try {
 564                                 info.updateData(bytes, offset, pos, size);
 565                         } catch (Throwable t) {
 566                                 throw new IllegalStateException(t);
 567                         } finally {
 568                                 info.releaseMutex();
 569                         }
 570
 571                 } catch (DatabaseException e) {
 572                         e.printStackTrace();
 573                 }
 574
 575         }
 576
 577     public void shutdown() {
 578         clusterLRU.shutdown();
 579         fileLRU.shutdown();
 580         streamLRU.shutdown();
 581         csLRU.shutdown();
 582     }
 583
 584 }