@utils.Synchronized
  def Compact(self, callback=None, timestamp=None):
    """Compacts versioned attributes into the collection stream.

    Versioned attributes come from the datastore sorted by the timestamp
    in the decreasing order. This is the opposite of what we want in
    the collection (as items in the collection should be in chronological
    order).

    Compact's implementation can handle very large collections that can't
    be reversed in memory. It reads them in batches, reverses every batch
    individually, and then reads batches back in the reversed order and
    write their contents to the collection stream.

    Args:
      callback: An optional function without arguments that gets called
                periodically while processing is done. Useful in flows
                that have to heartbeat.
      timestamp: Only items added before this timestamp will be compacted.

    Raises:
      RuntimeError: if problems are encountered when reading back temporary
                    saved data.

    Returns:
      Number of compacted results.
    """
    if not self.locked:
      raise aff4.LockError("Collection must be locked before compaction.")

    compacted_count = 0

    batches_urns = []
    current_batch = []

    # This timestamp will be used to delete attributes. We don't want
    # to delete anything that was added after we started the compaction.
    freeze_timestamp = timestamp or rdfvalue.RDFDatetime().Now()

    def UpdateIndex():
      seek_index = self.Get(self.Schema.SEEK_INDEX, SeekIndex())

      prev_index_pair = seek_index.checkpoints and seek_index.checkpoints[-1]
      if (not prev_index_pair or
          self.size - prev_index_pair.index_offset >= self.INDEX_INTERVAL):
        new_index_pair = SeekIndexPair(index_offset=self.size,
                                       byte_offset=self.fd.Tell())
        seek_index.checkpoints.Append(new_index_pair)
        self.Set(self.Schema.SEEK_INDEX, seek_index)

    def DeleteVersionedDataAndFlush():
      """Removes versioned attributes and flushes the stream."""
      data_store.DB.DeleteAttributes(self.urn, [self.Schema.DATA.predicate],
                                     end=freeze_timestamp,
                                     token=self.token, sync=True)
      if self.IsJournalingEnabled():
        journal_entry = self.Schema.COMPACTION_JOURNAL(compacted_count,
                                                       age=freeze_timestamp)
        attrs_to_set = {self.Schema.COMPACTION_JOURNAL: [journal_entry]}
        aff4.FACTORY.SetAttributes(self.urn, attrs_to_set, set(),
                                   add_child_index=False, sync=True,
                                   token=self.token)

      if self.Schema.DATA in self.synced_attributes:
        del self.synced_attributes[self.Schema.DATA]

      self.Flush(sync=True)

    def HeartBeat():
      """Update the lock lease if needed and call the callback."""
      lease_time = config_lib.CONFIG["Worker.compaction_lease_time"]
      if self.CheckLease() < 2 / lease_time:
        logging.info("%s: Extending compaction lease.", self.urn)
        self.UpdateLease(lease_time)
        stats.STATS.IncrementCounter("packed_collection_lease_extended")

      if callback:
        callback()

    HeartBeat()

    # We iterate over all versioned attributes. If we get more than
    # self.COMPACTION_BATCH_SIZE, we write the data to temporary
    # stream in the reversed order.
    for _, value, _ in data_store.DB.ResolvePrefix(
        self.urn, self.Schema.DATA.predicate, token=self.token,
        timestamp=(0, freeze_timestamp)):

      HeartBeat()

      current_batch.append(value)
      compacted_count += 1

      if len(current_batch) >= self.COMPACTION_BATCH_SIZE:
        batch_urn = rdfvalue.RDFURN("aff4:/tmp").Add(
            "%X" % utils.PRNG.GetULong())
        batches_urns.append(batch_urn)

        buf = cStringIO.StringIO()
        for data in reversed(current_batch):
          buf.write(struct.pack("<i", len(data)))
          buf.write(data)

        # We use AFF4Image to avoid serializing/deserializing data stored
        # in versioned attributes.
        with aff4.FACTORY.Create(batch_urn, "AFF4Image", mode="w",
                                 token=self.token) as batch_stream:
          batch_stream.Write(buf.getvalue())

        current_batch = []

    # If there are no versioned attributes, we have nothing to do.
    if not current_batch and not batches_urns:
      return 0

    # The last batch of results can be written to our collection's stream
    # immediately, because we have to reverse the order of all the data
    # stored in versioned attributes.
    if current_batch:
      buf = cStringIO.StringIO()
      for data in reversed(current_batch):
        buf.write(struct.pack("<i", len(data)))
        buf.write(data)

      self.fd.Seek(0, 2)
      self.fd.Write(buf.getvalue())
      self.stream_dirty = True
      self.size += len(current_batch)
      UpdateIndex()

      # If current_batch was the only available batch, just write everything
      # and return.
      if not batches_urns:
        DeleteVersionedDataAndFlush()
        return compacted_count

    batches = {}
    for batch in aff4.FACTORY.MultiOpen(batches_urns, aff4_type="AFF4Image",
                                        token=self.token):
      batches[batch.urn] = batch

    if len(batches_urns) != len(batches):
      raise RuntimeError("Internal inconsistency can't read back all the "
                         "temporary batches.")

    # We read all the temporary batches in reverse order (batches itself
    # were reversed when they were written).
    self.fd.Seek(0, 2)
    for batch_urn in reversed(batches_urns):
      batch = batches[batch_urn]

      HeartBeat()

      data = batch.Read(len(batch))
      self.fd.Write(data)
      self.stream_dirty = True
      self.size += self.COMPACTION_BATCH_SIZE
      UpdateIndex()

      aff4.FACTORY.Delete(batch_urn, token=self.token)

    DeleteVersionedDataAndFlush()

    # Update system-wide stats.
    stats.STATS.IncrementCounter("packed_collection_compacted",
                                 delta=compacted_count)

    return compacted_count
dataset/ETHPy150Open google/grr/grr/lib/aff4_objects/collects.py/PackedVersionedCollection.Compact/swapped operands of `/`