bundles/org.simantics.simulation/src/org/simantics/simulation/history/HistoryUtil.java

   1 /*******************************************************************************
   2  * Copyright (c) 2007, 2018 Association for Decentralized Information Management in
   3  * 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  *     Semantum Oy - archive implementation, gitlab simantics/platform#227
  12  *******************************************************************************/
  13 package org.simantics.simulation.history;
  14
  15 import java.io.EOFException;
  16 import java.io.IOException;
  17 import java.io.InputStream;
  18 import java.io.OutputStream;
  19 import java.nio.file.Path;
  20 import java.util.Arrays;
  21 import java.util.HashMap;
  22 import java.util.Map;
  23
  24 import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
  25 import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
  26 import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream;
  27 import org.eclipse.core.runtime.IStatus;
  28 import org.eclipse.core.runtime.Status;
  29 import org.simantics.databoard.Bindings;
  30 import org.simantics.databoard.accessor.StreamAccessor;
  31 import org.simantics.databoard.accessor.binary.BinaryObject;
  32 import org.simantics.databoard.accessor.error.AccessorException;
  33 import org.simantics.databoard.binding.ArrayBinding;
  34 import org.simantics.databoard.binding.Binding;
  35 import org.simantics.databoard.binding.RecordBinding;
  36 import org.simantics.databoard.binding.error.BindingConstructionException;
  37 import org.simantics.databoard.binding.error.BindingException;
  38 import org.simantics.databoard.binding.impl.ObjectArrayBinding;
  39 import org.simantics.databoard.binding.mutable.MutableVariant;
  40 import org.simantics.databoard.binding.mutable.Variant;
  41 import org.simantics.databoard.container.DataContainer;
  42 import org.simantics.databoard.container.DataContainers;
  43 import org.simantics.databoard.serialization.Serializer;
  44 import org.simantics.databoard.util.Bean;
  45 import org.simantics.databoard.util.binary.BinaryFile;
  46 import org.simantics.databoard.util.binary.OutputStreamWriteable;
  47 import org.simantics.databoard.util.binary.RandomAccessBinary;
  48 import org.simantics.db.ReadGraph;
  49 import org.simantics.db.Resource;
  50 import org.simantics.db.WriteGraph;
  51 import org.simantics.db.common.request.UniqueRead;
  52 import org.simantics.db.common.request.WriteRequest;
  53 import org.simantics.db.exception.DatabaseException;
  54 import org.simantics.db.exception.ServiceNotFoundException;
  55 import org.simantics.db.request.Read;
  56 import org.simantics.fastlz.FastLZ;
  57 import org.simantics.history.HistoryException;
  58 import org.simantics.history.HistoryManager;
  59 import org.simantics.history.ItemManager;
  60 import org.simantics.history.impl.CollectorImpl;
  61 import org.simantics.history.impl.CollectorState;
  62 import org.simantics.history.impl.CollectorState.VariableState;
  63 import org.simantics.history.util.Stream;
  64 import org.simantics.history.util.ValueBand;
  65 import org.simantics.history.util.WeightedMedian;
  66 import org.simantics.layer0.Layer0;
  67 import org.simantics.simulation.Activator;
  68 import org.simantics.simulation.ontology.HistoryResource;
  69 import org.simantics.simulation.ontology.SimulationResource;
  70 import org.simantics.utils.FileUtils;
  71
  72 import gnu.trove.map.TObjectLongMap;
  73 import gnu.trove.map.hash.TObjectLongHashMap;
  74 import net.jpountz.lz4.LZ4BlockInputStream;
  75 import net.jpountz.lz4.LZ4BlockOutputStream;
  76
  77 /**
  78  * @author Toni Kalajainen
  79  * @author Tuukka Lehtonen
  80  */
  81 public class HistoryUtil {
  82
  83         private static final boolean DEBUG = false;
  84         private static final boolean PROFILE = true;
  85
  86         public enum Compression {
  87                 FLZ,
  88                 LZ4,
  89                 NONE,
  90         }
  91
  92         private static final Compression USED_COMPRESSION = Compression.FLZ;
  93
  94         private static final String APPLICATION_X_TAR = "application/x-tar";
  95         private static final String APPLICATION_X_LZ4 = "application/x-lz4";
  96         private static final String APPLICATION_X_FLZ = "application/x-flz";
  97         private static final int ARCHIVE_VERSION_1 = 1; // TarArchiveInput/OutputStream
  98
  99         public static class ArchivedHistory {
 100                 public long totalSampleSize = 0L;
 101                 public TObjectLongMap<Bean> itemSizes = new TObjectLongHashMap<Bean>();
 102         }
 103
 104         /**
 105          * Create request that exports history to graph.
 106          *
 107          * If removeOldUnused is true, items that do not exist in source history
 108          * are deleted from graph.
 109          *
 110          * If item already exists in the graph, it is overwritten.
 111          *
 112          * @param history source history
 113          * @param g write graph
 114          * @param historyResource Instance of HIS.History to write data to
 115          * @param removeOldUnused
 116          * @param timeShift adjust time values by this value
 117          * @param from time
 118          * @param end time
 119          * @return WriteRequest
 120          */
 121         public static WriteRequest export(final HistoryManager history,
 122                         final Resource historyResource,
 123                         final boolean removeOldUnused,
 124                         final Double timeShift,
 125                         final double from, final double end)
 126         {
 127                 return new WriteRequest() {
 128                         public void perform(WriteGraph graph) throws DatabaseException {
 129                                 HistoryResource HIS = HistoryResource.getInstance(graph);
 130                                 Layer0 L0 = Layer0.getInstance(graph);
 131                                 long totalSampleSize = 0L;
 132                                 long s = System.nanoTime();
 133
 134                                 try {
 135                                         Bean[] items = history.getItems();
 136                                         if (PROFILE)
 137                                                 profile(null, "exporting " + items.length + " history items to database");
 138
 139                                         ItemManager im = new ItemManager( items );
 140                                         Map<String, Resource> oldResourceMap = new HashMap<String, Resource>();
 141                                         for (Resource oldItem : graph.getObjects(historyResource, L0.ConsistsOf))
 142                                         {
 143                                                 if (!graph.isInstanceOf(oldItem, HIS.History_Item)) continue;
 144                                                 Bean bean = graph.getPossibleRelatedValue(oldItem, HIS.History_Item_Info, Bindings.BEAN);
 145                                                 if ( bean == null ) continue;
 146                                                 String id = (String) bean.getField("id");
 147
 148                                                 if (!im.exists(id)) {
 149                                                         // There is item in graph that does not exist in source history
 150                                                         if ( removeOldUnused ) {
 151                                                                 graph.deny(oldItem);
 152                                                         }
 153                                                 } else {
 154                                                         // Item already exists, to-be-overwritten
 155                                                         oldResourceMap.put(id, oldItem);
 156                                                 }
 157                                         }
 158
 159                                         for (Bean newItem : im.values())
 160                                         {
 161                                                 //System.out.println("WRITING ITEM: " + newItem);
 162                                                 String id = (String) newItem.getField("id");
 163                                                 Resource historyItemResource = oldResourceMap.get(id);
 164                                                 Resource data = null;
 165
 166                                                 if ( historyItemResource==null ) {
 167                                                         historyItemResource = graph.newResource();
 168                                                         graph.claim(historyResource, L0.ConsistsOf, historyItemResource);
 169                                                         graph.claim(historyItemResource, L0.InstanceOf, HIS.History_Item);
 170                                                         graph.claimLiteral(historyItemResource, L0.HasName, id);
 171                                                 } else {
 172                                                         data = graph.getPossibleObject(historyItemResource, HIS.History_Item_Series);
 173                                                 }
 174
 175                                                 Variant v = new Variant(newItem.getBinding(), newItem);
 176                                                 graph.claimLiteral(historyItemResource, HIS.History_Item_Info, v, Bindings.VARIANT);
 177
 178                                                 if (data == null) {
 179                                                         data = graph.newResource();
 180                                                         graph.claim(data, L0.InstanceOf, L0.Variant);
 181                                                         graph.claim(historyItemResource, HIS.History_Item_Series, data);
 182                                                 } else {
 183                                                         graph.denyValue(data);
 184                                                 }
 185
 186
 187                                                 // Write stream
 188                                                 StreamAccessor sa = history.openStream(id, "r");
 189                                                 try {
 190                                                         //System.out.println("WRITING stream of size=" + sa.size());
 191                                                         RecordBinding sampleBinding = (RecordBinding) Bindings.getBeanBinding( sa.type().componentType );
 192                                                         Serializer serializer = Bindings.getSerializerUnchecked(sampleBinding);
 193                                                         Integer constantSampleSize = serializer.getConstantSize();
 194                                                         ValueBand vb = new ValueBand(sampleBinding);
 195                                                         Stream stream = new Stream(sa, sampleBinding);
 196                                                         ArrayBinding arrayBinding = new ObjectArrayBinding(sa.type(), sampleBinding);
 197                                                         Object array = null;
 198                                                         //System.out.println("WRITING stream: sample size=" + constantSampleSize);
 199
 200                                                         int count = 0;
 201                                                         readArray: {
 202                                                                 // Start index
 203                                                                 int startIndex = stream.binarySearch(Bindings.DOUBLE, from);
 204                                                                 if ( startIndex < -stream.count() ) break readArray;
 205                                                                 if ( startIndex<0 ) startIndex = -2-startIndex;
 206                                                                 if ( startIndex == -1 ) startIndex = 0;
 207
 208                                                                 // End index
 209                                                                 int endIndex = stream.binarySearch(Bindings.DOUBLE, end);
 210                                                                 if ( endIndex == -1 ) break readArray;
 211                                                                 if ( endIndex<0 ) endIndex = -1-endIndex;
 212                                                                 if ( endIndex == sa.size() ) endIndex = sa.size()-1;
 213                                                                 if ( endIndex<startIndex ) break readArray;
 214
 215                                                                 // Write sample count
 216                                                                 count = endIndex - startIndex + 1;
 217                                                                 array = arrayBinding.create(count);
 218                                                                 boolean hasTimeShift = timeShift!=null && timeShift != 0.0;
 219                                                                 //System.out.println("WRITING stream: count=" + count + ", datatype=" + arrayBinding.type());
 220                                                                 for (int i=0; i<count; i++) {
 221                                                                         Object sample = sa.get(i+startIndex, sampleBinding);
 222
 223                                                                         // Adjust time
 224                                                                         if ( hasTimeShift ) {
 225                                                                                 vb.setSample(sample);
 226                                                                                 if ( vb.hasTime() ) {
 227                                                                                         double newTime = vb.getTimeDouble() + timeShift;
 228                                                                                         vb.setTime(Bindings.DOUBLE, newTime);
 229                                                                                 }
 230                                                                                 if ( vb.hasEndTime() ) {
 231                                                                                         double newTime = vb.getEndTimeDouble() + timeShift;
 232                                                                                         vb.setEndTime(Bindings.DOUBLE, newTime);
 233                                                                                 }
 234                                                                         }
 235                                                                         //System.out.println("\t#" + i + ": sample=" + sample);
 236                                                                         arrayBinding.set(array, i, sample);
 237                                                                 }
 238                                                         }
 239                                                         if (array==null) array = arrayBinding.create();
 240                                                         Variant v2 = new Variant(arrayBinding, array);
 241                                                         graph.claimValue(data, v2, Bindings.VARIANT);
 242
 243                                                         if (constantSampleSize != null) {
 244                                                                 long itemSampleSize = ((long) count) * ((long) constantSampleSize);
 245                                                                 //System.out.println("item sample size: " + itemSampleSize);
 246                                                                 totalSampleSize += itemSampleSize;
 247                                                                 graph.claimLiteral(historyItemResource, HIS.History_Item_size, L0.Long, itemSampleSize, Bindings.LONG);
 248                                                         }
 249                                                 } catch (AccessorException e) {
 250                                                         throw new DatabaseException(e);
 251                                                 } finally {
 252                                                         try {
 253                                                                 sa.close();
 254                                                         } catch (AccessorException e) {
 255                                                         }
 256                                                 }
 257                                         }
 258
 259                                         graph.claimLiteral(historyResource, HIS.History_size, L0.Long, totalSampleSize, Bindings.LONG);
 260
 261                                         if (PROFILE)
 262                                                 profile(s, "exported " + items.length + " history items to database");
 263
 264                                 } catch (HistoryException e) {
 265                                         throw new DatabaseException( e );
 266                                 } catch (BindingException e) {
 267                                         throw new DatabaseException( e );
 268                                 } catch (ServiceNotFoundException e) {
 269                                         throw new DatabaseException( e );
 270                                 }
 271                         }
 272                 };
 273         }
 274
 275         /**
 276          * Create request that exports history to graph.
 277          *
 278          * If item already exists in the graph, it is overwritten.
 279          *
 280          * @param history
 281          *            source history
 282          * @param collectorState
 283          *            complete dump of the source history collector state or
 284          *            <code>null</code> to skip collector state saving
 285          * @param historyResource
 286          *            Instance of HIS.History to write data to
 287          * @param timeShift
 288          *            adjust time values by this value
 289          * @param from
 290          *            time
 291          * @param end
 292          *            time
 293          * @return WriteRequest
 294          */
 295         public static WriteRequest exportArchive(
 296                         final HistoryManager history,
 297                         final Bean collectorState,
 298                         final Resource historyResource,
 299                         final Double timeShift,
 300                         final double from,
 301                         final double end)
 302         {
 303                 return new WriteRequest() {
 304                         public void perform(WriteGraph graph) throws DatabaseException {
 305                                 HistoryResource HIS = HistoryResource.getInstance(graph);
 306                                 Layer0 L0 = Layer0.getInstance(graph);
 307
 308                                 long s = System.nanoTime();
 309                                 if (PROFILE)
 310                                         profile(null, "archiving history items to database");
 311
 312                                 // Remove all possibly existing old items in the history.
 313                                 for (Resource entity : graph.getObjects(historyResource, L0.ConsistsOf))
 314                                         if (graph.isInstanceOf(entity, HIS.History_Item))
 315                                                 graph.deny(historyResource, L0.ConsistsOf, L0.PartOf, entity);
 316
 317                                 // Create new literal for history archival
 318                                 //graph.deny(historyResource, HIS.History_archive);
 319                                 graph.denyValue(historyResource, HIS.History_archive);
 320                                 Resource archiveLiteral = graph.newResource();
 321                                 graph.claim(archiveLiteral, L0.InstanceOf, null, L0.ByteArray);
 322                                 graph.claim(historyResource, HIS.History_archive, archiveLiteral);
 323
 324                                 OutputStream closeable = null;
 325                                 try {
 326                                         RandomAccessBinary rab = graph.getRandomAccessBinary(archiveLiteral);
 327                                         rab.position(0);
 328                                         rab.skipBytes(4);
 329
 330                                         ArchivedHistory archive = exportCompressedArchive(history, collectorState, timeShift, from, end, rab);
 331
 332                                         rab.position(0L);
 333                                         rab.writeInt((int)(rab.length() - 4L));
 334
 335                                         graph.claimLiteral(historyResource, HIS.History_size, L0.Long, archive.totalSampleSize, Bindings.LONG);
 336
 337                                         if (PROFILE)
 338                                                 profile(s, "archived history items of size " + archive.totalSampleSize + " to database");
 339
 340                                 } catch (HistoryException e) {
 341                                         throw new DatabaseException( e );
 342                                 } catch (IOException e) {
 343                                         throw new DatabaseException( e );
 344                                 } finally {
 345                                         FileUtils.uncheckedClose(closeable);
 346                                 }
 347                         }
 348                 };
 349         }
 350
 351         public static ArchivedHistory exportCompressedArchive(
 352                         HistoryManager history,
 353                         Bean collectorState,
 354                         Double timeShift,
 355                         double from,
 356                         double end,
 357                         RandomAccessBinary rab)
 358                                         throws IOException, HistoryException
 359         {
 360                 return exportCompressedArchive(history, collectorState, timeShift, from, end, USED_COMPRESSION, rab);
 361         }
 362
 363         public static ArchivedHistory exportCompressedArchive(
 364                         HistoryManager history,
 365                         Bean collectorState,
 366                         Double timeShift,
 367                         double from,
 368                         double end,
 369                         Compression compression,
 370                         RandomAccessBinary rab)
 371                                         throws IOException, HistoryException
 372         {
 373                 switch (compression) {
 374                         case FLZ:
 375                                 DataContainers.writeHeader(rab, new DataContainer(APPLICATION_X_FLZ, ARCHIVE_VERSION_1, null));
 376                                 return exportArchive(history, collectorState, timeShift, from, end, FastLZ.write(new RABOutputStream(rab)));
 377                         case LZ4:
 378                                 DataContainers.writeHeader(rab, new DataContainer(APPLICATION_X_LZ4, ARCHIVE_VERSION_1, null));
 379                                 return exportArchive(history, collectorState, timeShift, from, end, new LZ4BlockOutputStream(new RABOutputStream(rab)));
 380                         case NONE:
 381                                 DataContainers.writeHeader(rab, new DataContainer(APPLICATION_X_TAR, ARCHIVE_VERSION_1, null));
 382                                 return exportArchive(history, collectorState, timeShift, from, end, new RABOutputStream(rab));
 383                         default:
 384                                 throw new UnsupportedOperationException("Unsupported compression: " + compression);
 385                 }
 386         }
 387
 388         private static ArchivedHistory exportArchive(
 389                         HistoryManager history,
 390                         Bean collectorState,
 391                         Double timeShift,
 392                         double from,
 393                         double end,
 394                         OutputStream out)
 395                                         throws IOException, HistoryException
 396         {
 397                 try (OutputStream os = out) {
 398                         ArchivedHistory archive = exportArchive( os, history, collectorState, timeShift, from, end );
 399                         os.flush();
 400                         return archive;
 401                 }
 402         }
 403
 404         /**
 405          * Store the specified history manager's contents into a single archive
 406          * file.
 407          *
 408          * @param outputStream
 409          *            the output stream where the history archive is written
 410          * @param history
 411          *            source history
 412          * @param collectorState
 413          *            complete state of the history collector at the time of saving
 414          *            or <code>null</code> to not save any state
 415          * @param timeShift
 416          *            adjust time values by this value
 417          * @param from
 418          *            time
 419          * @param end
 420          *            time
 421          * @return ArchivedHistory instance describing the created archive
 422          * @throws HistoryException
 423          *             when there's a problem with reading the history data from the
 424          *             history work area
 425          * @throws IOException
 426          *             when there's a problem writing the resulting history archive
 427          *             file
 428          */
 429         private static ArchivedHistory exportArchive(
 430                         OutputStream outputStream,
 431                         HistoryManager history,
 432                         Bean collectorState,
 433                         Double timeShift,
 434                         double from,
 435                         double end)
 436                                         throws IOException, HistoryException
 437         {
 438                 ArchivedHistory result = new ArchivedHistory();
 439
 440                 Serializer beanSerializer = Bindings.getSerializerUnchecked(Bindings.BEAN);
 441                 boolean hasTimeShift = timeShift != null && timeShift != 0.0;
 442
 443                 TarArchiveOutputStream out = new TarArchiveOutputStream(outputStream);
 444                 OutputStreamWriteable dataOutput = new OutputStreamWriteable(out);
 445
 446                 try {
 447                         if (collectorState != null) {
 448                                 // Write complete collector state into the archive
 449                                 if (DEBUG)
 450                                         System.out.println("WRITING collector state: " + collectorState);
 451                                 byte[] serializedCollectorState = beanSerializer.serialize( collectorState );
 452                                 putNextEntry(out, "state", serializedCollectorState.length);
 453                                 out.write(serializedCollectorState);
 454                                 out.closeArchiveEntry();
 455                         }
 456
 457                         ItemManager im = new ItemManager( history.getItems() );
 458
 459                         for (Bean item : im.values()) {
 460                                 if (DEBUG)
 461                                         System.out.println("STORING ITEM: " + item);
 462                                 String id = (String) item.getField("id");
 463
 464                                 // Write data stream metadata
 465                                 byte[] metadata = beanSerializer.serialize( item );
 466                                 putNextEntry(out, id + ".txt", metadata.length);
 467                                 out.write(metadata);
 468                                 out.closeArchiveEntry();
 469                                 if (DEBUG)
 470                                         System.out.println("SERIALIZED ITEM " + id + ": " + Arrays.toString(metadata));
 471
 472
 473                                 // Write data stream as a separate entry
 474                                 StreamAccessor sa = history.openStream(id, "r");
 475                                 try {
 476                                         if (DEBUG)
 477                                                 System.out.println("STREAM SIZE=" + sa.size());
 478                                         RecordBinding sampleBinding = (RecordBinding) Bindings.getBeanBinding( sa.type().componentType );
 479                                         Serializer sampleSerializer = Bindings.getSerializerUnchecked(sampleBinding);
 480                                         Integer constantSampleSize = sampleSerializer.getConstantSize();
 481
 482                                         if (constantSampleSize == null) {
 483                                                 // Variable length samples - no support for now.
 484                                                 System.err.println("Skipping item " + item + " in history archival due to variable length samples not being supported at the moment.");
 485                                                 continue;
 486                                         } else {
 487                                                 ValueBand vb = new ValueBand(sampleBinding);
 488                                                 Stream stream = new Stream(sa, sampleBinding);
 489                                                 if (DEBUG)
 490                                                         System.out.println("WRITING stream: sample size=" + constantSampleSize);
 491
 492                                                 readArray: {
 493                                                         // Start index
 494                                                         int startIndex = stream.binarySearch(Bindings.DOUBLE, from);
 495                                                         if ( startIndex < -stream.count() ) break readArray;
 496                                                         if ( startIndex<0 ) startIndex = -2-startIndex;
 497                                                         if ( startIndex == -1 ) startIndex = 0;
 498
 499                                                         // End index
 500                                                         int endIndex = stream.binarySearch(Bindings.DOUBLE, end);
 501                                                         if ( endIndex == -1 ) break readArray;
 502                                                         if ( endIndex<0 ) endIndex = -1-endIndex;
 503                                                         if ( endIndex == sa.size() ) endIndex = sa.size()-1;
 504                                                         if ( endIndex<startIndex ) break readArray;
 505
 506                                                         // Written sample count
 507                                                         int count = endIndex - startIndex + 1;
 508                                                         if (DEBUG)
 509                                                                 System.out.println("WRITTEN sample count=" + count);
 510
 511                                                         long savedSamplesSize = ((long) count) * ((long) constantSampleSize);
 512                                                         if (DEBUG)
 513                                                                 System.out.println("saved samples size in bytes: " + savedSamplesSize);
 514                                                         putNextEntry(out, id + ".data", savedSamplesSize);
 515
 516                                                         for (int i=0; i<count; i++) {
 517                                                                 Object sample = sa.get(i+startIndex, sampleBinding);
 518
 519                                                                 // Adjust time
 520                                                                 if ( hasTimeShift ) {
 521                                                                         vb.setSample(sample);
 522                                                                         if ( vb.hasTime() ) {
 523                                                                                 double newTime = vb.getTimeDouble() + timeShift;
 524                                                                                 vb.setTime(Bindings.DOUBLE, newTime);
 525                                                                         }
 526                                                                         if ( vb.hasEndTime() ) {
 527                                                                                 double newTime = vb.getEndTimeDouble() + timeShift;
 528                                                                                 vb.setEndTime(Bindings.DOUBLE, newTime);
 529                                                                         }
 530                                                                 }
 531                                                                 //System.out.println("\t#" + i + ": sample=" + sample);
 532                                                                 sampleSerializer.serialize(dataOutput, sample);
 533                                                         }
 534                                                         out.closeArchiveEntry();
 535
 536                                                         result.itemSizes.put(item, savedSamplesSize);
 537                                                         result.totalSampleSize += savedSamplesSize;
 538                                                 }
 539                                         }
 540                                 } catch (AccessorException e) {
 541                                         throw new IOException(e);
 542                                 } finally {
 543                                         try {
 544                                                 sa.close();
 545                                         } catch (AccessorException e) {
 546                                         }
 547                                 }
 548
 549                         }
 550
 551                         return result;
 552                 } catch (BindingException e) {
 553                         throw new HistoryException(e);
 554                 } finally {
 555                         out.finish();
 556                 }
 557         }
 558
 559         /**
 560          * Import all history items from graph into a history manager.
 561          *
 562          * @param r history resource or initial condition resource source HIS.History resource
 563          * @param history destination history
 564          * @return read request that always returns a HistoryImportResult instance
 565          */
 566         public static Read<HistoryImportResult> importHistory(final Resource r, final HistoryManager history)
 567         {
 568                 return new UniqueRead<HistoryImportResult>() {
 569                         @Override
 570                         public HistoryImportResult perform(ReadGraph graph) throws DatabaseException {
 571                                 HistoryImportResult result = new HistoryImportResult();
 572
 573                                 HistoryResource HIS = HistoryResource.getInstance(graph);
 574                                 SimulationResource SIM = SimulationResource.getInstance(graph);
 575                                 Resource historyResource = r;
 576
 577                                 if (!graph.isInstanceOf(historyResource, HIS.History))
 578                                         historyResource = graph.getPossibleObject(r, SIM.State_History);
 579                                 if (historyResource == null)
 580                                         return result;
 581                                 if (!graph.isInstanceOf(historyResource, HIS.History))
 582                                         return result;
 583
 584                                 long s = System.nanoTime();
 585
 586                                 Resource archive = graph.getPossibleObject(historyResource, HIS.History_archive);
 587                                 if (archive == null) {
 588                                         if (PROFILE)
 589                                                 profile(null, "importing history items from old database format to disk workarea");
 590                                         importHistoryItems(graph, historyResource, history);
 591                                 } else {
 592                                         try {
 593                                                 if (PROFILE)
 594                                                         profile(null, "importing history items from archived format to disk workarea");
 595                                                 importHistoryArchive(graph, archive, history, result);
 596                                         } catch (IOException e) {
 597                                                 throw new DatabaseException(e);
 598                                         } catch (HistoryException e) {
 599                                                 throw new DatabaseException(e);
 600                                         }
 601                                 }
 602
 603                                 if (PROFILE)
 604                                         profile(s, "imported history items from database to disk workarea");
 605
 606                                 return result;
 607                         }
 608                 };
 609         }
 610
 611         /**
 612          * Import all history items from graph into a history manager.
 613          *
 614          * @param r history resource or initial condition resource source HIS.History resource
 615          * @param history destination history
 616          * @return <code>true</code> if successful
 617          */
 618         private static boolean importHistoryItems(ReadGraph graph, final Resource historyResource, final HistoryManager history) throws DatabaseException
 619         {
 620                 HistoryResource HIS = HistoryResource.getInstance(graph);
 621                 Layer0 L0 = Layer0.getInstance(graph);
 622
 623                 try {
 624                         for (Resource oldItem : graph.getObjects(historyResource, L0.ConsistsOf)) {
 625                                 if (!graph.isInstanceOf(oldItem, HIS.History_Item))
 626                                         continue;
 627
 628                                 Bean bean = graph.getRelatedValue(oldItem, HIS.History_Item_Info, Bindings.BEAN);
 629                                 String id = (String) bean.getFieldUnchecked("id");
 630                                 Object array = graph.getRelatedValue(oldItem, HIS.History_Item_Series, Bindings.BEAN);
 631                                 Binding arrayBinding = Bindings.getBinding( array.getClass() );
 632                                 history.modify(bean);
 633                                 StreamAccessor sa = history.openStream(id, "rw");
 634                                 try {
 635                                         sa.setValue(arrayBinding, array);
 636                                 } finally {
 637                                         try {
 638                                                 sa.close();
 639                                         } catch (AccessorException e) {
 640                                         }
 641                                 }
 642                         }
 643
 644                         return true;
 645                 } catch (AccessorException e) {
 646                         throw new DatabaseException(e);
 647                 } catch (BindingConstructionException e) {
 648                         throw new DatabaseException(e);
 649                 } catch (HistoryException e) {
 650                         throw new DatabaseException(e);
 651                 }
 652         }
 653
 654         /**
 655          * @param history
 656          * @param path
 657          * @return
 658          * @throws HistoryException
 659          * @throws IOException
 660          */
 661         public static HistoryImportResult importHistoryArchive(HistoryManager history, Path path) throws IOException, HistoryException {
 662                 HistoryImportResult result = new HistoryImportResult();
 663                 try (RandomAccessBinary rab = new BinaryFile(path.toFile())) {
 664                         importHistoryArchive(history, rab, result);
 665                         return result;
 666                 }
 667         }
 668
 669         /**
 670          * Import all history items from database archive into a history manager.
 671          *
 672          * @param graph
 673          * @param historyResource
 674          * @param archive
 675          * @param history
 676          * @return <code>true</code> if successful
 677          * @throws DatabaseException
 678          */
 679         private static boolean importHistoryArchive(ReadGraph graph, Resource archive, HistoryManager history, HistoryImportResult result) throws DatabaseException, IOException, HistoryException {
 680                 RandomAccessBinary rab = graph.getRandomAccessBinary(archive);
 681                 // This is here because the data is encoded as an L0.ByteArray
 682                 // in the database which means the first integer describes the
 683                 // length of the byte array. We skip that length data here.
 684                 rab.position(4);
 685                 return importHistoryArchive(history, rab, result);
 686         }
 687
 688         /**
 689          * Import all history items from database archive into a history manager.
 690          *
 691          * @param graph
 692          * @param historyResource
 693          * @param archive
 694          * @param history
 695          * @return <code>true</code> if successful
 696          * @throws DatabaseException
 697          */
 698         private static boolean importHistoryArchive(HistoryManager history, RandomAccessBinary rab, HistoryImportResult result)
 699                         throws IOException, HistoryException
 700         {
 701                 DataContainer dc = DataContainers.readHeader(rab);
 702
 703                 if (DEBUG)
 704                         System.out.println("COMPRESSED HISTORY DATA SIZE: " + (rab.length() - rab.position()));
 705
 706                 if (dc.version == ARCHIVE_VERSION_1) {
 707                         if (APPLICATION_X_LZ4.equals(dc.format)) {
 708                                 return extractHistoryArchiveInputStream(history, new LZ4BlockInputStream(new RABInputStream(rab)), result);
 709                         } else if (APPLICATION_X_FLZ.equals(dc.format)) {
 710                                 return extractHistoryArchiveInputStream(history, FastLZ.read(new RABInputStream(rab)), result);
 711                         } else if (APPLICATION_X_TAR.equals(dc.format)) {
 712                                 return extractHistoryArchiveInputStream(history, new RABInputStream(rab), result);
 713                         }
 714                         throw new UnsupportedOperationException("Unsupported format " + dc.format + " and version " + dc.version);
 715                 } else {
 716                         throw new UnsupportedOperationException("Unsupported history archive version " + dc.version + " with format " + dc.format);
 717                 }
 718         }
 719
 720         private static boolean extractHistoryArchiveInputStream(HistoryManager history, InputStream in, HistoryImportResult result) throws IOException, HistoryException {
 721                 try (InputStream _in = in) {
 722                         return extractHistoryArchiveTar(history, in, result);
 723                 }
 724         }
 725
 726         /**
 727          * Import all history items from graph into a history manager
 728          *
 729          * @param graph database access
 730          * @param history destination history
 731          * @param rab the archive to extract from
 732          * @return <code>true</code> if successful
 733          * @throws DatabaseException
 734          */
 735         private static boolean extractHistoryArchiveTar(HistoryManager history, InputStream in, HistoryImportResult result)
 736                         throws IOException, HistoryException
 737         {
 738                 // tar is not closed on purpose because we do not want to close RandomAccessBinary rab.
 739                 TarArchiveInputStream tar = new TarArchiveInputStream(in);
 740
 741                 Serializer beanSerializer = Bindings.getSerializerUnchecked(Bindings.BEAN);
 742
 743                 Bean lastItem = null;
 744                 String lastItemId = null;
 745                 while (true) {
 746                         TarArchiveEntry entry = tar.getNextTarEntry();
 747                         if (entry == null)
 748                                 break;
 749
 750                         String name = entry.getName();
 751                         boolean state = name.equals("state");
 752                         boolean metadata = name.endsWith(".txt");
 753                         boolean data = name.endsWith(".data");
 754
 755                         if (state) {
 756                                 if (result != null) {
 757                                         byte[] st = new byte[(int) entry.getSize()];
 758                                         tar.read(st);
 759                                         result.collectorState = (Bean) beanSerializer.deserialize(st);
 760                                         if (DEBUG)
 761                                                 System.out.println("READ collector state: " + result.collectorState);
 762                                 } else {
 763                                         tar.skip(entry.getSize());
 764                                 }
 765                         } else if (metadata) {
 766                                 byte[] md = new byte[(int) entry.getSize()];
 767                                 tar.read(md);
 768                                 if (DEBUG) {
 769                                         System.out.println("READING Item metadata: " + name);
 770                                         System.out.println("SERIALIZED ITEM " + name.replace(".txt", "") + ": " + Arrays.toString(md));
 771                                 }
 772
 773                                 try {
 774                                         Bean bean = (Bean) beanSerializer.deserialize(md);
 775                                         if (DEBUG)
 776                                                 System.out.println("READ Item metadata: " + bean);
 777                                         history.modify(bean);
 778                                         lastItem = bean;
 779                                         lastItemId = (String) bean.getFieldUnchecked("id");
 780                                 } catch (ClassFormatError e) {
 781                                         // This is here because LZ4BlockInput/OutputStream seemed to
 782                                         // be causing weird deserialization errors to occur for
 783                                         // single data items when trying to deserialize beans. The
 784                                         // reason being that the serialized data that was read only
 785                                         // contains the first 512 bytes of the data. After that all
 786                                         // data is zeros, which causes things like this to happen in
 787                                         // deserialization:
 788                                         //signature: rsstzzzzzzzze_b7b532e9
 789                                         //component(o): id = String
 790                                         //component(1): variableId = String
 791                                         //component(2): format = Temp1
 792                                         //annotation: @org.simantics.databoard.annotations.Union(value=[class org.simantics.databoard.type.BooleanType, class org.simantics.databoard.type.ByteType, class org.simantics.databoard.type.IntegerType, class org.simantics.databoard.type.LongType, class org.simantics.databoard.type.FloatType, class org.simantics.databoard.type.DoubleType, class org.simantics.databoard.type.StringType, class org.simantics.databoard.type.RecordType, class org.simantics.databoard.type.ArrayType, class org.simantics.databoard.type.MapType, class org.simantics.databoard.type.OptionalType, class org.simantics.databoard.type.UnionType, class org.simantics.databoard.type.VariantType])
 793                                         //component(3):  = Boolean
 794                                         //component(4):  = Boolean
 795                                         //component(5):  = Boolean
 796                                         //component(6):  = Boolean
 797                                         //component(7):  = Boolean
 798                                         //component(8):  = Boolean
 799                                         //component(9):  = Boolean
 800                                         //component(10):  = Boolean
 801                                         //
 802                                         // For this reason we've switched the default compression to FastLZ
 803                                         // for now. This is reflected in the input format also.
 804
 805                                         Activator.getDefault().getLog().log(
 806                                                         new Status(IStatus.ERROR, Activator.PLUGIN_ID, "History item " + name + " metadata deserialization failed.", e));;
 807                                 }
 808                         } else if (data && lastItem != null) {
 809                                 StreamAccessor sa = history.openStream(lastItemId, "rw");
 810                                 try {
 811                                         if (sa instanceof BinaryObject) {
 812                                                 BinaryObject bo = (BinaryObject) sa;
 813                                                 RandomAccessBinary output = bo.getBinary();
 814                                                 output.position(0L);
 815                                                 output.setLength(0L);
 816                                                 long copiedBytes = FileUtils.copy(tar, new RABOutputStream(output), entry.getSize());
 817                                                 if (copiedBytes != entry.getSize()) {
 818                                                         System.err.println("WARNING: item " + lastItemId + ", extracted " + copiedBytes + " bytes while TAR entry said the size to be " + entry.getSize() + " bytes");
 819                                                 }
 820                                         } else {
 821                                                 System.err.println("Skipping item " + lastItemId + " in history import due to StreamAccessor not being an instanceof BinaryObject. StreamAccessor=" + sa);
 822                                         }
 823                                 } finally {
 824                                         try {
 825                                                 sa.close();
 826                                                 lastItem = null;
 827                                                 lastItemId = null;
 828                                         } catch (AccessorException e) {
 829                                         }
 830                                 }
 831                         }
 832                 }
 833
 834                 return true;
 835         }
 836
 837         /**
 838          * Get request that clears history
 839          * @return cleaning request
 840          */
 841         public static WriteRequest clear(final Resource history) {
 842                 return new WriteRequest() {
 843                         @Override
 844                         public void perform(WriteGraph graph) throws DatabaseException {
 845                                 HistoryResource HIS = HistoryResource.getInstance(graph);
 846                                 Layer0 L0 = Layer0.getInstance(graph);
 847
 848                                 // Separate items format
 849                                 for (Resource oldItem : graph.getObjects(history, L0.ConsistsOf))
 850                                 {
 851                                         if (!graph.isInstanceOf(oldItem, HIS.History_Item)) continue;
 852                                         Resource info = graph.getPossibleObject(oldItem, HIS.History_Item);
 853                                         if (info!=null) {
 854                                                 graph.deny(oldItem, HIS.History_Item_Info);
 855                                                 graph.deny(info);
 856                                         }
 857                                         Resource data = graph.getPossibleObject(oldItem, HIS.History_Item_Series);
 858                                         if (data!=null) {
 859                                                 graph.deny(oldItem, HIS.History_Item_Series);
 860                                                 graph.deny(data);
 861                                         }
 862                                         graph.deny(history, L0.ConsistsOf, oldItem);
 863                                         graph.deny(oldItem);
 864                                 }
 865
 866                                 // Archived format
 867                                 graph.denyValue(history, HIS.History_archive);
 868                         }
 869                 };
 870         }
 871
 872
 873         private static void putNextEntry(TarArchiveOutputStream out, String entryName, long size) throws IOException {
 874                 TarArchiveEntry entry = new TarArchiveEntry(entryName);
 875                 entry.setSize(size);
 876                 out.putArchiveEntry(entry);
 877         }
 878
 879         /**
 880          * Shall not close the underlying RandomAccessBinary.
 881          */
 882         private static class RABOutputStream extends OutputStream {
 883
 884                 private final RandomAccessBinary output;
 885
 886                 public RABOutputStream(RandomAccessBinary output) {
 887                         this.output = output;
 888                 }
 889
 890                 @Override
 891                 public void write(byte[] b, int off, int len) throws IOException {
 892                         output.write(b, off, len);
 893                 }
 894
 895                 @Override
 896                 public void write(int b) throws IOException {
 897                         output.write(b);
 898                 }
 899
 900         }
 901
 902         /**
 903          * Shall not close the underlying RandomAccessBinary.
 904          */
 905         private static class RABInputStream extends InputStream {
 906
 907                 private final RandomAccessBinary input;
 908
 909                 public RABInputStream(RandomAccessBinary input) {
 910                         this.input = input;
 911                 }
 912
 913                 @Override
 914                 public int available() throws IOException {
 915                         long avail = input.length() - input.position();
 916                         if ((avail & 0xffffffff00000000L) != 0)
 917                                 throw new IOException("Number of available bytes truncated in long->int conversion: " + avail);
 918                         return (int) avail;
 919                 }
 920
 921                 @Override
 922                 public int read() throws IOException {
 923                         try {
 924                                 return input.readUnsignedByte();
 925                         } catch (EOFException e) {
 926                                 return -1;
 927                         }
 928                 }
 929
 930                 @Override
 931                 public int read(byte[] b, int off, int len) throws IOException {
 932                         long available = input.length() - input.position();
 933                         if (available == 0)
 934                                 return -1;
 935                         int l = (int) Math.min(available, (long) len);
 936                         input.readFully(b, off, l);
 937                         return l;
 938                 }
 939
 940                 @Override
 941                 public long skip(long n) throws IOException {
 942                         long count = 0;
 943                         while (count < n) {
 944                                 int l = (int) Math.min(n, (long) Integer.MAX_VALUE);
 945                                 count += input.skipBytes(l);
 946                         }
 947                         return count;
 948                 }
 949
 950         }
 951
 952         private static long profile(Long t1, String string) {
 953                 if (PROFILE) {
 954                         long t2 = System.nanoTime();
 955                         long delta = t1 == null ? 0 : t2-t1;
 956                         System.out.println("[" + HistoryUtil.class.getSimpleName() + "] PROFILE: " + string + (t1 == null ? "" : " in " + (((double)delta)*1e-6) + " ms"));
 957                         return t2;
 958                 }
 959                 return 0L;
 960         }
 961
 962         public static CollectorState truncateHistory(double toBeforeTime, HistoryManager history, CollectorState state) throws AccessorException, BindingException, HistoryException {
 963                 Double t = toBeforeTime;
 964                 Binding timeBinding = null;
 965
 966                 Bean[] items = history.getItems();
 967                 //System.out.println("truncating all samples after t=" + toBeforeTime + " for " + items.length + " history items");
 968
 969                 for (Bean item : items) {
 970                         String id = (String) item.getField("id");
 971                         StreamAccessor sa = history.openStream(id, "rw");
 972                         try {
 973                                 Stream s = new Stream(sa);
 974                                 timeBinding = s.timeBinding;
 975                                 int currentSize = sa.size();
 976                                 int index = s.binarySearch(timeBinding, toBeforeTime);
 977                                 int newSize = truncationSize(index);
 978                                 if (newSize < currentSize) {
 979                                         //System.out.println("truncating item: " + item + " from size " + currentSize + " to " + newSize);
 980                                         sa.setSize(newSize);
 981
 982                                         if (state != null) {
 983                                                 Object prevTime  = newSize > 0 ? s.getItemTime(s.timeBinding, newSize - 1) : null;
 984                                                 Bean prevSample = newSize > 0 ? (Bean) sa.get(newSize - 1, s.sampleBinding) : null;
 985                                                 Object prevValue = prevSample != null ? prevSample.getField(s.valueIndex) : null;
 986                                                 boolean isNan = isNaN(prevValue);
 987
 988                                                 VariableState vs = state.values.get(id);
 989                                                 if (vs != null && vs.value != null && prevValue != null) {
 990                                                         vs.value.setValue(vs.value.getBinding(), prevValue);
 991                                                         vs.isValid = true;
 992                                                         vs.isNan = isNan;
 993                                                 }
 994
 995                                                 CollectorState.Item is = state.itemStates.get(id);
 996                                                 if (is != null) {
 997                                                         is.firstTime = toTime(prevTime);
 998                                                         is.firstValue = toValue(s.valueBinding, prevValue);
 999                                                         is.currentTime = is.firstTime;
1000                                                         is.currentValue = toValue(s.valueBinding, prevValue);
1001                                                         is.isNaN = isNan;
1002                                                         is.isValid = prevValue != null;
1003                                                         is.sum = Double.NaN;
1004                                                         is.count = 0;
1005                                                         is.ooDeadband = false;
1006                                                         is.firstDisabledTime = Double.NaN;
1007                                                         is.lastDisabledTime = Double.NaN;
1008                                                         is.median = new WeightedMedian( CollectorImpl.MEDIAN_LIMIT );
1009                                                 }
1010                                         }
1011                                 }
1012                         } finally {
1013                                 sa.close();
1014                         }
1015                 }
1016
1017                 if (timeBinding != null && state != null) {
1018                         state.time.setValue(timeBinding, t);
1019                         state.dT = 1.0;
1020                 }
1021
1022                 return state;
1023         }
1024
1025         private static double toTime(Object time) {
1026                 return time instanceof Number ? ((Number) time).doubleValue() : Double.NaN;
1027         }
1028
1029         private static MutableVariant toValue(Binding valueBinding, Object value) {
1030                 return new MutableVariant(valueBinding, value != null ? value : valueBinding.createDefaultUnchecked());
1031         }
1032
1033         private static boolean isNaN(Object value) {
1034                 return value instanceof Number ? Double.isNaN(((Number) value).doubleValue()) : false;
1035         }
1036
1037         private static int truncationSize(int binarySearchResult) {
1038                 return binarySearchResult >= 0 ? binarySearchResult + 1 : (-binarySearchResult - 1);
1039         }
1040
1041 }