jgit/org.eclipse.jgit/src/org/eclipse/jgit/internal/storage/dfs/DfsPackParser.java

/*
 * Copyright (C) 2011, Google Inc.
 * and other copyright owners as documented in the project's IP log.
 *
 * This program and the accompanying materials are made available
 * under the terms of the Eclipse Distribution License v1.0 which
 * accompanies this distribution, is reproduced below, and is
 * available at http://www.eclipse.org/org/documents/edl-v10.php
 *
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or
 * without modification, are permitted provided that the following
 * conditions are met:
 *
 * - Redistributions of source code must retain the above copyright
 *   notice, this list of conditions and the following disclaimer.
 *
 * - Redistributions in binary form must reproduce the above
 *   copyright notice, this list of conditions and the following
 *   disclaimer in the documentation and/or other materials provided
 *   with the distribution.
 *
 * - Neither the name of the Eclipse Foundation, Inc. nor the
 *   names of its contributors may be used to endorse or promote
 *   products derived from this software without specific prior
 *   written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND
 * CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
 * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
 * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
 * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

package org.eclipse.jgit.internal.storage.dfs;

import static org.eclipse.jgit.internal.storage.pack.PackExt.PACK;

import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.util.Collections;
import java.util.List;
import java.util.zip.CRC32;
import java.util.zip.Deflater;

import org.eclipse.jgit.internal.storage.file.PackIndex;
import org.eclipse.jgit.internal.storage.file.PackLock;
import org.eclipse.jgit.lib.AnyObjectId;
import org.eclipse.jgit.lib.Constants;
import org.eclipse.jgit.lib.ProgressMonitor;
import org.eclipse.jgit.transport.PackParser;
import org.eclipse.jgit.transport.PackedObjectInfo;

/**
 * Parses a pack stream into the DFS, by creating a new pack and index.
 */
public class DfsPackParser extends PackParser {
	private final DfsObjDatabase objdb;

	private final DfsInserter objins;

	/** CRC-32 computation for objects that are appended onto the pack. */
	private final CRC32 crc;

	/** Running SHA-1 of the entire pack stream. */
	private final MessageDigest packDigest;

	/** Block size to use when caching data for read back. */
	private int blockSize;

	/** Current end of the pack file. */
	private long packEnd;

	/** Checksum of the entire pack file. */
	private byte[] packHash;

	/** Compresses delta bases when completing a thin pack. */
	private Deflater def;

	/** True if the pack is an empty pack. */
	private boolean isEmptyPack;

	/** Name of the pack file, computed in {@link #onPackHeader(long)}. */
	private DfsPackDescription packDsc;

	/** Key used during delta resolution reading delta chains. */
	private DfsStreamKey packKey;

	/** If the index was small enough, the entire index after writing. */
	private PackIndex packIndex;

	/** Stream to the DFS storage, opened during {@link #onPackHeader(long)}. */
	private DfsOutputStream out;

	/** Data being written that has not yet been cached. */
	private byte[] currBuf;
	private long currPos; // Position of currBuf in the file.
	private int currEnd; // Position inside of currBuf to append to next.

	/** Cache the chunks were stored into or get read back from. */
	private DfsBlockCache blockCache;

	/** Cached block that is being read. */
	private long readPos;
	private DfsBlock readBlock;

	/**
	 * Initialize a new pack parser.
	 *
	 * @param db
	 *            database the objects will be imported into.
	 * @param ins
	 *            inserter the parser will use to help it inject the objects.
	 * @param in
	 *            the stream to parse.
	 */
	protected DfsPackParser(DfsObjDatabase db, DfsInserter ins, InputStream in) {
		super(db, in);
		this.objdb = db;
		this.objins = ins;
		this.crc = new CRC32();
		this.packDigest = Constants.newMessageDigest();
	}

	/** {@inheritDoc} */
	@Override
	public PackLock parse(ProgressMonitor receiving, ProgressMonitor resolving)
			throws IOException {
		boolean rollback = true;
		try {
			blockCache = DfsBlockCache.getInstance();
			super.parse(receiving, resolving);
			if (isEmptyPack)
				return null;
			buffer(packHash, 0, packHash.length);
			if (currEnd != 0)
				flushBlock();
			out.close();
			out = null;
			currBuf = null;
			readBlock = null;
			packDsc.addFileExt(PACK);
			packDsc.setFileSize(PACK, packEnd);
			packDsc.setBlockSize(PACK, blockSize);

			writePackIndex();
			objdb.commitPack(Collections.singletonList(packDsc), null);
			rollback = false;

			DfsPackFile p = new DfsPackFile(blockCache, packDsc);
			p.setBlockSize(blockSize);
			if (packIndex != null)
				p.setPackIndex(packIndex);

			objdb.addPack(p);

			return null;
		} finally {
			blockCache = null;
			currBuf = null;
			readBlock = null;

			if (def != null) {
				def.end();
				def = null;
			}

			if (out != null) {
				try {
					out.close();
				} catch (IOException err) {
					// Ignore a close error, rollbackPack is also important.
				}
				out = null;
			}

			if (rollback && packDsc != null) {
				try {
					objdb.rollbackPack(Collections.singletonList(packDsc));
				} finally {
					packDsc = null;
				}
			}
		}
	}

	/**
	 * Get description of the imported pack, if one was made.
	 *
	 * @return description of the imported pack, if one was made.
	 */
	public DfsPackDescription getPackDescription() {
		return packDsc;
	}

	/** {@inheritDoc} */
	@Override
	protected void onPackHeader(long objectCount) throws IOException {
		if (objectCount == 0) {
			isEmptyPack = true;
			currBuf = new byte[256];
			return;
		}

		packDsc = objdb.newPack(DfsObjDatabase.PackSource.RECEIVE);
		out = objdb.writeFile(packDsc, PACK);
		packKey = packDsc.getStreamKey(PACK);

		int size = out.blockSize();
		if (size <= 0)
			size = blockCache.getBlockSize();
		else if (size < blockCache.getBlockSize())
			size = (blockCache.getBlockSize() / size) * size;
		blockSize = size;
		currBuf = new byte[blockSize];
	}

	/** {@inheritDoc} */
	@Override
	protected void onBeginWholeObject(long streamPosition, int type,
			long inflatedSize) throws IOException {
		crc.reset();
	}

	/** {@inheritDoc} */
	@Override
	protected void onEndWholeObject(PackedObjectInfo info) throws IOException {
		info.setCRC((int) crc.getValue());
	}

	/** {@inheritDoc} */
	@Override
	protected void onBeginOfsDelta(long streamPosition,
			long baseStreamPosition, long inflatedSize) throws IOException {
		crc.reset();
	}

	/** {@inheritDoc} */
	@Override
	protected void onBeginRefDelta(long streamPosition, AnyObjectId baseId,
			long inflatedSize) throws IOException {
		crc.reset();
	}

	/** {@inheritDoc} */
	@Override
	protected UnresolvedDelta onEndDelta() throws IOException {
		UnresolvedDelta delta = new UnresolvedDelta();
		delta.setCRC((int) crc.getValue());
		return delta;
	}

	/** {@inheritDoc} */
	@Override
	protected void onInflatedObjectData(PackedObjectInfo obj, int typeCode,
			byte[] data) throws IOException {
		// DfsPackParser ignores this event.
	}

	/** {@inheritDoc} */
	@Override
	protected void onObjectHeader(Source src, byte[] raw, int pos, int len)
			throws IOException {
		crc.update(raw, pos, len);
	}

	/** {@inheritDoc} */
	@Override
	protected void onObjectData(Source src, byte[] raw, int pos, int len)
			throws IOException {
		crc.update(raw, pos, len);
	}

	/** {@inheritDoc} */
	@Override
	protected void onStoreStream(byte[] raw, int pos, int len)
			throws IOException {
		buffer(raw, pos, len);
		packDigest.update(raw, pos, len);
	}

	private void buffer(byte[] raw, int pos, int len) throws IOException {
		while (0 < len) {
			int n = Math.min(len, currBuf.length - currEnd);
			if (n == 0) {
				DfsBlock v = flushBlock();
				currBuf = new byte[blockSize];
				currEnd = 0;
				currPos += v.size();
				continue;
			}

			System.arraycopy(raw, pos, currBuf, currEnd, n);
			pos += n;
			len -= n;
			currEnd += n;
			packEnd += n;
		}
	}

	private DfsBlock flushBlock() throws IOException {
		if (isEmptyPack)
			throw new IOException(DfsText.get().willNotStoreEmptyPack);

		out.write(currBuf, 0, currEnd);

		byte[] buf;
		if (currEnd == currBuf.length) {
			buf = currBuf;
		} else {
			buf = new byte[currEnd];
			System.arraycopy(currBuf, 0, buf, 0, currEnd);
		}

		DfsBlock v = new DfsBlock(packKey, currPos, buf);
		readBlock = v;
		blockCache.put(v);
		return v;
	}

	/** {@inheritDoc} */
	@Override
	protected void onPackFooter(byte[] hash) throws IOException {
		// The base class will validate the original hash matches
		// what the stream has stored at the end. We are called
		// only if the hash was good. Save it in case there are no
		// missing bases to append.
		packHash = hash;
	}

	/** {@inheritDoc} */
	@Override
	protected ObjectTypeAndSize seekDatabase(PackedObjectInfo obj,
			ObjectTypeAndSize info) throws IOException {
		readPos = obj.getOffset();
		crc.reset();
		return readObjectHeader(info);
	}

	/** {@inheritDoc} */
	@Override
	protected ObjectTypeAndSize seekDatabase(UnresolvedDelta delta,
			ObjectTypeAndSize info) throws IOException {
		readPos = delta.getOffset();
		crc.reset();
		return readObjectHeader(info);
	}

	/** {@inheritDoc} */
	@Override
	protected int readDatabase(byte[] dst, int pos, int cnt) throws IOException {
		if (cnt == 0)
			return 0;

		if (currPos <= readPos) {
			// Requested read is still buffered. Copy direct from buffer.
			int p = (int) (readPos - currPos);
			int n = Math.min(cnt, currEnd - p);
			if (n == 0)
				throw new EOFException();
			System.arraycopy(currBuf, p, dst, pos, n);
			readPos += n;
			return n;
		}

		if (readBlock == null || !readBlock.contains(packKey, readPos)) {
			long start = toBlockStart(readPos);
			readBlock = blockCache.get(packKey, start);
			if (readBlock == null) {
				int size = (int) Math.min(blockSize, packEnd - start);
				byte[] buf = new byte[size];
				if (read(start, buf, 0, size) != size)
					throw new EOFException();
				readBlock = new DfsBlock(packKey, start, buf);
				blockCache.put(readBlock);
			}
		}

		int n = readBlock.copy(readPos, dst, pos, cnt);
		readPos += n;
		return n;
	}

	private int read(long pos, byte[] dst, int off, int len) throws IOException {
		if (len == 0)
			return 0;

		int cnt = 0;
		while (0 < len) {
			int r = out.read(pos, ByteBuffer.wrap(dst, off, len));
			if (r <= 0)
				break;
			pos += r;
			off += r;
			len -= r;
			cnt += r;
		}
		return cnt != 0 ? cnt : -1;
	}

	private long toBlockStart(long pos) {
		return (pos / blockSize) * blockSize;
	}

	/** {@inheritDoc} */
	@Override
	protected boolean checkCRC(int oldCRC) {
		return oldCRC == (int) crc.getValue();
	}

	/** {@inheritDoc} */
	@Override
	protected boolean onAppendBase(final int typeCode, final byte[] data,
			final PackedObjectInfo info) throws IOException {
		info.setOffset(packEnd);

		final byte[] buf = buffer();
		int sz = data.length;
		int len = 0;
		buf[len++] = (byte) ((typeCode << 4) | (sz & 15));
		sz >>>= 4;
		while (sz > 0) {
			buf[len - 1] |= (byte) 0x80;
			buf[len++] = (byte) (sz & 0x7f);
			sz >>>= 7;
		}

		packDigest.update(buf, 0, len);
		crc.reset();
		crc.update(buf, 0, len);
		buffer(buf, 0, len);

		if (def == null)
			def = new Deflater(Deflater.DEFAULT_COMPRESSION, false);
		else
			def.reset();
		def.setInput(data);
		def.finish();

		while (!def.finished()) {
			len = def.deflate(buf);
			packDigest.update(buf, 0, len);
			crc.update(buf, 0, len);
			buffer(buf, 0, len);
		}

		info.setCRC((int) crc.getValue());
		return true;
	}

	/** {@inheritDoc} */
	@Override
	protected void onEndThinPack() throws IOException {
		// Normally when a thin pack is closed the pack header gets
		// updated to reflect the actual object count. This is not going
		// to be possible on most DFS backends, so instead we allow
		// the header to have an incorrect count, but we do change the
		// trailing digest to be correct.
		packHash = packDigest.digest();
	}

	private void writePackIndex() throws IOException {
		List<PackedObjectInfo> list = getSortedObjectList(null /* by ObjectId */);
		packIndex = objins.writePackIndex(packDsc, packHash, list);
	}
}