You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

1034 lines
38KB

  1. # Copyright (C) 2010, 2011 Sebastian Thiel (byronimo@gmail.com) and contributors
  2. #
  3. # This module is part of GitDB and is released under
  4. # the New BSD License: http://www.opensource.org/licenses/bsd-license.php
  5. """Contains PackIndexFile and PackFile implementations"""
  6. import zlib
  7. from gitdb.exc import (
  8. BadObject,
  9. AmbiguousObjectName,
  10. UnsupportedOperation,
  11. ParseError
  12. )
  13. from gitdb.util import (
  14. mman,
  15. LazyMixin,
  16. unpack_from,
  17. bin_to_hex,
  18. byte_ord,
  19. )
  20. from gitdb.fun import (
  21. create_pack_object_header,
  22. pack_object_header_info,
  23. is_equal_canonical_sha,
  24. type_id_to_type_map,
  25. write_object,
  26. stream_copy,
  27. chunk_size,
  28. delta_types,
  29. OFS_DELTA,
  30. REF_DELTA,
  31. msb_size
  32. )
  33. try:
  34. from gitdb_speedups._perf import PackIndexFile_sha_to_index
  35. except ImportError:
  36. pass
  37. # END try c module
  38. from gitdb.base import ( # Amazing !
  39. OInfo,
  40. OStream,
  41. OPackInfo,
  42. OPackStream,
  43. ODeltaStream,
  44. ODeltaPackInfo,
  45. ODeltaPackStream,
  46. )
  47. from gitdb.stream import (
  48. DecompressMemMapReader,
  49. DeltaApplyReader,
  50. Sha1Writer,
  51. NullStream,
  52. FlexibleSha1Writer
  53. )
  54. from struct import pack
  55. from binascii import crc32
  56. from gitdb.const import NULL_BYTE
  57. from gitdb.utils.compat import (
  58. izip,
  59. buffer,
  60. xrange,
  61. to_bytes
  62. )
  63. import tempfile
  64. import array
  65. import os
  66. import sys
  67. __all__ = ('PackIndexFile', 'PackFile', 'PackEntity')
  68. #{ Utilities
  69. def pack_object_at(cursor, offset, as_stream):
  70. """
  71. :return: Tuple(abs_data_offset, PackInfo|PackStream)
  72. an object of the correct type according to the type_id of the object.
  73. If as_stream is True, the object will contain a stream, allowing the
  74. data to be read decompressed.
  75. :param data: random accessible data containing all required information
  76. :parma offset: offset in to the data at which the object information is located
  77. :param as_stream: if True, a stream object will be returned that can read
  78. the data, otherwise you receive an info object only"""
  79. data = cursor.use_region(offset).buffer()
  80. type_id, uncomp_size, data_rela_offset = pack_object_header_info(data)
  81. total_rela_offset = None # set later, actual offset until data stream begins
  82. delta_info = None
  83. # OFFSET DELTA
  84. if type_id == OFS_DELTA:
  85. i = data_rela_offset
  86. c = byte_ord(data[i])
  87. i += 1
  88. delta_offset = c & 0x7f
  89. while c & 0x80:
  90. c = byte_ord(data[i])
  91. i += 1
  92. delta_offset += 1
  93. delta_offset = (delta_offset << 7) + (c & 0x7f)
  94. # END character loop
  95. delta_info = delta_offset
  96. total_rela_offset = i
  97. # REF DELTA
  98. elif type_id == REF_DELTA:
  99. total_rela_offset = data_rela_offset + 20
  100. delta_info = data[data_rela_offset:total_rela_offset]
  101. # BASE OBJECT
  102. else:
  103. # assume its a base object
  104. total_rela_offset = data_rela_offset
  105. # END handle type id
  106. abs_data_offset = offset + total_rela_offset
  107. if as_stream:
  108. stream = DecompressMemMapReader(buffer(data, total_rela_offset), False, uncomp_size)
  109. if delta_info is None:
  110. return abs_data_offset, OPackStream(offset, type_id, uncomp_size, stream)
  111. else:
  112. return abs_data_offset, ODeltaPackStream(offset, type_id, uncomp_size, delta_info, stream)
  113. else:
  114. if delta_info is None:
  115. return abs_data_offset, OPackInfo(offset, type_id, uncomp_size)
  116. else:
  117. return abs_data_offset, ODeltaPackInfo(offset, type_id, uncomp_size, delta_info)
  118. # END handle info
  119. # END handle stream
  120. def write_stream_to_pack(read, write, zstream, base_crc=None):
  121. """Copy a stream as read from read function, zip it, and write the result.
  122. Count the number of written bytes and return it
  123. :param base_crc: if not None, the crc will be the base for all compressed data
  124. we consecutively write and generate a crc32 from. If None, no crc will be generated
  125. :return: tuple(no bytes read, no bytes written, crc32) crc might be 0 if base_crc
  126. was false"""
  127. br = 0 # bytes read
  128. bw = 0 # bytes written
  129. want_crc = base_crc is not None
  130. crc = 0
  131. if want_crc:
  132. crc = base_crc
  133. # END initialize crc
  134. while True:
  135. chunk = read(chunk_size)
  136. br += len(chunk)
  137. compressed = zstream.compress(chunk)
  138. bw += len(compressed)
  139. write(compressed) # cannot assume return value
  140. if want_crc:
  141. crc = crc32(compressed, crc)
  142. # END handle crc
  143. if len(chunk) != chunk_size:
  144. break
  145. # END copy loop
  146. compressed = zstream.flush()
  147. bw += len(compressed)
  148. write(compressed)
  149. if want_crc:
  150. crc = crc32(compressed, crc)
  151. # END handle crc
  152. return (br, bw, crc)
  153. #} END utilities
  154. class IndexWriter(object):
  155. """Utility to cache index information, allowing to write all information later
  156. in one go to the given stream
  157. **Note:** currently only writes v2 indices"""
  158. __slots__ = '_objs'
  159. def __init__(self):
  160. self._objs = list()
  161. def append(self, binsha, crc, offset):
  162. """Append one piece of object information"""
  163. self._objs.append((binsha, crc, offset))
  164. def write(self, pack_sha, write):
  165. """Write the index file using the given write method
  166. :param pack_sha: binary sha over the whole pack that we index
  167. :return: sha1 binary sha over all index file contents"""
  168. # sort for sha1 hash
  169. self._objs.sort(key=lambda o: o[0])
  170. sha_writer = FlexibleSha1Writer(write)
  171. sha_write = sha_writer.write
  172. sha_write(PackIndexFile.index_v2_signature)
  173. sha_write(pack(">L", PackIndexFile.index_version_default))
  174. # fanout
  175. tmplist = list((0,) * 256) # fanout or list with 64 bit offsets
  176. for t in self._objs:
  177. tmplist[byte_ord(t[0][0])] += 1
  178. # END prepare fanout
  179. for i in xrange(255):
  180. v = tmplist[i]
  181. sha_write(pack('>L', v))
  182. tmplist[i + 1] += v
  183. # END write each fanout entry
  184. sha_write(pack('>L', tmplist[255]))
  185. # sha1 ordered
  186. # save calls, that is push them into c
  187. sha_write(b''.join(t[0] for t in self._objs))
  188. # crc32
  189. for t in self._objs:
  190. sha_write(pack('>L', t[1] & 0xffffffff))
  191. # END for each crc
  192. tmplist = list()
  193. # offset 32
  194. for t in self._objs:
  195. ofs = t[2]
  196. if ofs > 0x7fffffff:
  197. tmplist.append(ofs)
  198. ofs = 0x80000000 + len(tmplist) - 1
  199. # END hande 64 bit offsets
  200. sha_write(pack('>L', ofs & 0xffffffff))
  201. # END for each offset
  202. # offset 64
  203. for ofs in tmplist:
  204. sha_write(pack(">Q", ofs))
  205. # END for each offset
  206. # trailer
  207. assert(len(pack_sha) == 20)
  208. sha_write(pack_sha)
  209. sha = sha_writer.sha(as_hex=False)
  210. write(sha)
  211. return sha
  212. class PackIndexFile(LazyMixin):
  213. """A pack index provides offsets into the corresponding pack, allowing to find
  214. locations for offsets faster."""
  215. # Dont use slots as we dynamically bind functions for each version, need a dict for this
  216. # The slots you see here are just to keep track of our instance variables
  217. # __slots__ = ('_indexpath', '_fanout_table', '_cursor', '_version',
  218. # '_sha_list_offset', '_crc_list_offset', '_pack_offset', '_pack_64_offset')
  219. # used in v2 indices
  220. _sha_list_offset = 8 + 1024
  221. index_v2_signature = b'\xfftOc'
  222. index_version_default = 2
  223. def __init__(self, indexpath):
  224. super(PackIndexFile, self).__init__()
  225. self._indexpath = indexpath
  226. def close(self):
  227. mman.force_map_handle_removal_win(self._indexpath)
  228. self._cursor = None
  229. def _set_cache_(self, attr):
  230. if attr == "_packfile_checksum":
  231. self._packfile_checksum = self._cursor.map()[-40:-20]
  232. elif attr == "_packfile_checksum":
  233. self._packfile_checksum = self._cursor.map()[-20:]
  234. elif attr == "_cursor":
  235. # Note: We don't lock the file when reading as we cannot be sure
  236. # that we can actually write to the location - it could be a read-only
  237. # alternate for instance
  238. self._cursor = mman.make_cursor(self._indexpath).use_region()
  239. # We will assume that the index will always fully fit into memory !
  240. if mman.window_size() > 0 and self._cursor.file_size() > mman.window_size():
  241. raise AssertionError("The index file at %s is too large to fit into a mapped window (%i > %i). This is a limitation of the implementation" % (
  242. self._indexpath, self._cursor.file_size(), mman.window_size()))
  243. # END assert window size
  244. else:
  245. # now its time to initialize everything - if we are here, someone wants
  246. # to access the fanout table or related properties
  247. # CHECK VERSION
  248. mmap = self._cursor.map()
  249. self._version = (mmap[:4] == self.index_v2_signature and 2) or 1
  250. if self._version == 2:
  251. version_id = unpack_from(">L", mmap, 4)[0]
  252. assert version_id == self._version, "Unsupported index version: %i" % version_id
  253. # END assert version
  254. # SETUP FUNCTIONS
  255. # setup our functions according to the actual version
  256. for fname in ('entry', 'offset', 'sha', 'crc'):
  257. setattr(self, fname, getattr(self, "_%s_v%i" % (fname, self._version)))
  258. # END for each function to initialize
  259. # INITIALIZE DATA
  260. # byte offset is 8 if version is 2, 0 otherwise
  261. self._initialize()
  262. # END handle attributes
  263. #{ Access V1
  264. def _entry_v1(self, i):
  265. """:return: tuple(offset, binsha, 0)"""
  266. return unpack_from(">L20s", self._cursor.map(), 1024 + i * 24) + (0, )
  267. def _offset_v1(self, i):
  268. """see ``_offset_v2``"""
  269. return unpack_from(">L", self._cursor.map(), 1024 + i * 24)[0]
  270. def _sha_v1(self, i):
  271. """see ``_sha_v2``"""
  272. base = 1024 + (i * 24) + 4
  273. return self._cursor.map()[base:base + 20]
  274. def _crc_v1(self, i):
  275. """unsupported"""
  276. return 0
  277. #} END access V1
  278. #{ Access V2
  279. def _entry_v2(self, i):
  280. """:return: tuple(offset, binsha, crc)"""
  281. return (self._offset_v2(i), self._sha_v2(i), self._crc_v2(i))
  282. def _offset_v2(self, i):
  283. """:return: 32 or 64 byte offset into pack files. 64 byte offsets will only
  284. be returned if the pack is larger than 4 GiB, or 2^32"""
  285. offset = unpack_from(">L", self._cursor.map(), self._pack_offset + i * 4)[0]
  286. # if the high-bit is set, this indicates that we have to lookup the offset
  287. # in the 64 bit region of the file. The current offset ( lower 31 bits )
  288. # are the index into it
  289. if offset & 0x80000000:
  290. offset = unpack_from(">Q", self._cursor.map(), self._pack_64_offset + (offset & ~0x80000000) * 8)[0]
  291. # END handle 64 bit offset
  292. return offset
  293. def _sha_v2(self, i):
  294. """:return: sha at the given index of this file index instance"""
  295. base = self._sha_list_offset + i * 20
  296. return self._cursor.map()[base:base + 20]
  297. def _crc_v2(self, i):
  298. """:return: 4 bytes crc for the object at index i"""
  299. return unpack_from(">L", self._cursor.map(), self._crc_list_offset + i * 4)[0]
  300. #} END access V2
  301. #{ Initialization
  302. def _initialize(self):
  303. """initialize base data"""
  304. self._fanout_table = self._read_fanout((self._version == 2) * 8)
  305. if self._version == 2:
  306. self._crc_list_offset = self._sha_list_offset + self.size() * 20
  307. self._pack_offset = self._crc_list_offset + self.size() * 4
  308. self._pack_64_offset = self._pack_offset + self.size() * 4
  309. # END setup base
  310. def _read_fanout(self, byte_offset):
  311. """Generate a fanout table from our data"""
  312. d = self._cursor.map()
  313. out = list()
  314. append = out.append
  315. for i in xrange(256):
  316. append(unpack_from('>L', d, byte_offset + i * 4)[0])
  317. # END for each entry
  318. return out
  319. #} END initialization
  320. #{ Properties
  321. def version(self):
  322. return self._version
  323. def size(self):
  324. """:return: amount of objects referred to by this index"""
  325. return self._fanout_table[255]
  326. def path(self):
  327. """:return: path to the packindexfile"""
  328. return self._indexpath
  329. def packfile_checksum(self):
  330. """:return: 20 byte sha representing the sha1 hash of the pack file"""
  331. return self._cursor.map()[-40:-20]
  332. def indexfile_checksum(self):
  333. """:return: 20 byte sha representing the sha1 hash of this index file"""
  334. return self._cursor.map()[-20:]
  335. def offsets(self):
  336. """:return: sequence of all offsets in the order in which they were written
  337. **Note:** return value can be random accessed, but may be immmutable"""
  338. if self._version == 2:
  339. # read stream to array, convert to tuple
  340. a = array.array('I') # 4 byte unsigned int, long are 8 byte on 64 bit it appears
  341. a.fromstring(buffer(self._cursor.map(), self._pack_offset, self._pack_64_offset - self._pack_offset))
  342. # networkbyteorder to something array likes more
  343. if sys.byteorder == 'little':
  344. a.byteswap()
  345. return a
  346. else:
  347. return tuple(self.offset(index) for index in xrange(self.size()))
  348. # END handle version
  349. def sha_to_index(self, sha):
  350. """
  351. :return: index usable with the ``offset`` or ``entry`` method, or None
  352. if the sha was not found in this pack index
  353. :param sha: 20 byte sha to lookup"""
  354. first_byte = byte_ord(sha[0])
  355. get_sha = self.sha
  356. lo = 0 # lower index, the left bound of the bisection
  357. if first_byte != 0:
  358. lo = self._fanout_table[first_byte - 1]
  359. hi = self._fanout_table[first_byte] # the upper, right bound of the bisection
  360. # bisect until we have the sha
  361. while lo < hi:
  362. mid = (lo + hi) // 2
  363. mid_sha = get_sha(mid)
  364. if sha < mid_sha:
  365. hi = mid
  366. elif sha == mid_sha:
  367. return mid
  368. else:
  369. lo = mid + 1
  370. # END handle midpoint
  371. # END bisect
  372. return None
  373. def partial_sha_to_index(self, partial_bin_sha, canonical_length):
  374. """
  375. :return: index as in `sha_to_index` or None if the sha was not found in this
  376. index file
  377. :param partial_bin_sha: an at least two bytes of a partial binary sha as bytes
  378. :param canonical_length: length of the original hexadecimal representation of the
  379. given partial binary sha
  380. :raise AmbiguousObjectName:"""
  381. if len(partial_bin_sha) < 2:
  382. raise ValueError("Require at least 2 bytes of partial sha")
  383. assert isinstance(partial_bin_sha, bytes), "partial_bin_sha must be bytes"
  384. first_byte = byte_ord(partial_bin_sha[0])
  385. get_sha = self.sha
  386. lo = 0 # lower index, the left bound of the bisection
  387. if first_byte != 0:
  388. lo = self._fanout_table[first_byte - 1]
  389. hi = self._fanout_table[first_byte] # the upper, right bound of the bisection
  390. # fill the partial to full 20 bytes
  391. filled_sha = partial_bin_sha + NULL_BYTE * (20 - len(partial_bin_sha))
  392. # find lowest
  393. while lo < hi:
  394. mid = (lo + hi) // 2
  395. mid_sha = get_sha(mid)
  396. if filled_sha < mid_sha:
  397. hi = mid
  398. elif filled_sha == mid_sha:
  399. # perfect match
  400. lo = mid
  401. break
  402. else:
  403. lo = mid + 1
  404. # END handle midpoint
  405. # END bisect
  406. if lo < self.size():
  407. cur_sha = get_sha(lo)
  408. if is_equal_canonical_sha(canonical_length, partial_bin_sha, cur_sha):
  409. next_sha = None
  410. if lo + 1 < self.size():
  411. next_sha = get_sha(lo + 1)
  412. if next_sha and next_sha == cur_sha:
  413. raise AmbiguousObjectName(partial_bin_sha)
  414. return lo
  415. # END if we have a match
  416. # END if we found something
  417. return None
  418. if 'PackIndexFile_sha_to_index' in globals():
  419. # NOTE: Its just about 25% faster, the major bottleneck might be the attr
  420. # accesses
  421. def sha_to_index(self, sha):
  422. return PackIndexFile_sha_to_index(self, sha)
  423. # END redefine heavy-hitter with c version
  424. #} END properties
  425. class PackFile(LazyMixin):
  426. """A pack is a file written according to the Version 2 for git packs
  427. As we currently use memory maps, it could be assumed that the maximum size of
  428. packs therefor is 32 bit on 32 bit systems. On 64 bit systems, this should be
  429. fine though.
  430. **Note:** at some point, this might be implemented using streams as well, or
  431. streams are an alternate path in the case memory maps cannot be created
  432. for some reason - one clearly doesn't want to read 10GB at once in that
  433. case"""
  434. __slots__ = ('_packpath', '_cursor', '_size', '_version')
  435. pack_signature = 0x5041434b # 'PACK'
  436. pack_version_default = 2
  437. # offset into our data at which the first object starts
  438. first_object_offset = 3 * 4 # header bytes
  439. footer_size = 20 # final sha
  440. def __init__(self, packpath):
  441. self._packpath = packpath
  442. def close(self):
  443. mman.force_map_handle_removal_win(self._packpath)
  444. self._cursor = None
  445. def _set_cache_(self, attr):
  446. # we fill the whole cache, whichever attribute gets queried first
  447. self._cursor = mman.make_cursor(self._packpath).use_region()
  448. # read the header information
  449. type_id, self._version, self._size = unpack_from(">LLL", self._cursor.map(), 0)
  450. # TODO: figure out whether we should better keep the lock, or maybe
  451. # add a .keep file instead ?
  452. if type_id != self.pack_signature:
  453. raise ParseError("Invalid pack signature: %i" % type_id)
  454. def _iter_objects(self, start_offset, as_stream=True):
  455. """Handle the actual iteration of objects within this pack"""
  456. c = self._cursor
  457. content_size = c.file_size() - self.footer_size
  458. cur_offset = start_offset or self.first_object_offset
  459. null = NullStream()
  460. while cur_offset < content_size:
  461. data_offset, ostream = pack_object_at(c, cur_offset, True)
  462. # scrub the stream to the end - this decompresses the object, but yields
  463. # the amount of compressed bytes we need to get to the next offset
  464. stream_copy(ostream.read, null.write, ostream.size, chunk_size)
  465. assert ostream.stream._br == ostream.size
  466. cur_offset += (data_offset - ostream.pack_offset) + ostream.stream.compressed_bytes_read()
  467. # if a stream is requested, reset it beforehand
  468. # Otherwise return the Stream object directly, its derived from the
  469. # info object
  470. if as_stream:
  471. ostream.stream.seek(0)
  472. yield ostream
  473. # END until we have read everything
  474. #{ Pack Information
  475. def size(self):
  476. """:return: The amount of objects stored in this pack"""
  477. return self._size
  478. def version(self):
  479. """:return: the version of this pack"""
  480. return self._version
  481. def data(self):
  482. """
  483. :return: read-only data of this pack. It provides random access and usually
  484. is a memory map.
  485. :note: This method is unsafe as it returns a window into a file which might be larger than than the actual window size"""
  486. # can use map as we are starting at offset 0. Otherwise we would have to use buffer()
  487. return self._cursor.use_region().map()
  488. def checksum(self):
  489. """:return: 20 byte sha1 hash on all object sha's contained in this file"""
  490. return self._cursor.use_region(self._cursor.file_size() - 20).buffer()[:]
  491. def path(self):
  492. """:return: path to the packfile"""
  493. return self._packpath
  494. #} END pack information
  495. #{ Pack Specific
  496. def collect_streams(self, offset):
  497. """
  498. :return: list of pack streams which are required to build the object
  499. at the given offset. The first entry of the list is the object at offset,
  500. the last one is either a full object, or a REF_Delta stream. The latter
  501. type needs its reference object to be locked up in an ODB to form a valid
  502. delta chain.
  503. If the object at offset is no delta, the size of the list is 1.
  504. :param offset: specifies the first byte of the object within this pack"""
  505. out = list()
  506. c = self._cursor
  507. while True:
  508. ostream = pack_object_at(c, offset, True)[1]
  509. out.append(ostream)
  510. if ostream.type_id == OFS_DELTA:
  511. offset = ostream.pack_offset - ostream.delta_info
  512. else:
  513. # the only thing we can lookup are OFFSET deltas. Everything
  514. # else is either an object, or a ref delta, in the latter
  515. # case someone else has to find it
  516. break
  517. # END handle type
  518. # END while chaining streams
  519. return out
  520. #} END pack specific
  521. #{ Read-Database like Interface
  522. def info(self, offset):
  523. """Retrieve information about the object at the given file-absolute offset
  524. :param offset: byte offset
  525. :return: OPackInfo instance, the actual type differs depending on the type_id attribute"""
  526. return pack_object_at(self._cursor, offset or self.first_object_offset, False)[1]
  527. def stream(self, offset):
  528. """Retrieve an object at the given file-relative offset as stream along with its information
  529. :param offset: byte offset
  530. :return: OPackStream instance, the actual type differs depending on the type_id attribute"""
  531. return pack_object_at(self._cursor, offset or self.first_object_offset, True)[1]
  532. def stream_iter(self, start_offset=0):
  533. """
  534. :return: iterator yielding OPackStream compatible instances, allowing
  535. to access the data in the pack directly.
  536. :param start_offset: offset to the first object to iterate. If 0, iteration
  537. starts at the very first object in the pack.
  538. **Note:** Iterating a pack directly is costly as the datastream has to be decompressed
  539. to determine the bounds between the objects"""
  540. return self._iter_objects(start_offset, as_stream=True)
  541. #} END Read-Database like Interface
  542. class PackEntity(LazyMixin):
  543. """Combines the PackIndexFile and the PackFile into one, allowing the
  544. actual objects to be resolved and iterated"""
  545. __slots__ = ('_index', # our index file
  546. '_pack', # our pack file
  547. '_offset_map' # on demand dict mapping one offset to the next consecutive one
  548. )
  549. IndexFileCls = PackIndexFile
  550. PackFileCls = PackFile
  551. def __init__(self, pack_or_index_path):
  552. """Initialize ourselves with the path to the respective pack or index file"""
  553. basename, ext = os.path.splitext(pack_or_index_path)
  554. self._index = self.IndexFileCls("%s.idx" % basename) # PackIndexFile instance
  555. self._pack = self.PackFileCls("%s.pack" % basename) # corresponding PackFile instance
  556. def close(self):
  557. self._index.close()
  558. self._pack.close()
  559. def _set_cache_(self, attr):
  560. # currently this can only be _offset_map
  561. # TODO: make this a simple sorted offset array which can be bisected
  562. # to find the respective entry, from which we can take a +1 easily
  563. # This might be slower, but should also be much lighter in memory !
  564. offsets_sorted = sorted(self._index.offsets())
  565. last_offset = len(self._pack.data()) - self._pack.footer_size
  566. assert offsets_sorted, "Cannot handle empty indices"
  567. offset_map = None
  568. if len(offsets_sorted) == 1:
  569. offset_map = {offsets_sorted[0]: last_offset}
  570. else:
  571. iter_offsets = iter(offsets_sorted)
  572. iter_offsets_plus_one = iter(offsets_sorted)
  573. next(iter_offsets_plus_one)
  574. consecutive = izip(iter_offsets, iter_offsets_plus_one)
  575. offset_map = dict(consecutive)
  576. # the last offset is not yet set
  577. offset_map[offsets_sorted[-1]] = last_offset
  578. # END handle offset amount
  579. self._offset_map = offset_map
  580. def _sha_to_index(self, sha):
  581. """:return: index for the given sha, or raise"""
  582. index = self._index.sha_to_index(sha)
  583. if index is None:
  584. raise BadObject(sha)
  585. return index
  586. def _iter_objects(self, as_stream):
  587. """Iterate over all objects in our index and yield their OInfo or OStream instences"""
  588. _sha = self._index.sha
  589. _object = self._object
  590. for index in xrange(self._index.size()):
  591. yield _object(_sha(index), as_stream, index)
  592. # END for each index
  593. def _object(self, sha, as_stream, index=-1):
  594. """:return: OInfo or OStream object providing information about the given sha
  595. :param index: if not -1, its assumed to be the sha's index in the IndexFile"""
  596. # its a little bit redundant here, but it needs to be efficient
  597. if index < 0:
  598. index = self._sha_to_index(sha)
  599. if sha is None:
  600. sha = self._index.sha(index)
  601. # END assure sha is present ( in output )
  602. offset = self._index.offset(index)
  603. type_id, uncomp_size, data_rela_offset = pack_object_header_info(self._pack._cursor.use_region(offset).buffer())
  604. if as_stream:
  605. if type_id not in delta_types:
  606. packstream = self._pack.stream(offset)
  607. return OStream(sha, packstream.type, packstream.size, packstream.stream)
  608. # END handle non-deltas
  609. # produce a delta stream containing all info
  610. # To prevent it from applying the deltas when querying the size,
  611. # we extract it from the delta stream ourselves
  612. streams = self.collect_streams_at_offset(offset)
  613. dstream = DeltaApplyReader.new(streams)
  614. return ODeltaStream(sha, dstream.type, None, dstream)
  615. else:
  616. if type_id not in delta_types:
  617. return OInfo(sha, type_id_to_type_map[type_id], uncomp_size)
  618. # END handle non-deltas
  619. # deltas are a little tougher - unpack the first bytes to obtain
  620. # the actual target size, as opposed to the size of the delta data
  621. streams = self.collect_streams_at_offset(offset)
  622. buf = streams[0].read(512)
  623. offset, src_size = msb_size(buf)
  624. offset, target_size = msb_size(buf, offset)
  625. # collect the streams to obtain the actual object type
  626. if streams[-1].type_id in delta_types:
  627. raise BadObject(sha, "Could not resolve delta object")
  628. return OInfo(sha, streams[-1].type, target_size)
  629. # END handle stream
  630. #{ Read-Database like Interface
  631. def info(self, sha):
  632. """Retrieve information about the object identified by the given sha
  633. :param sha: 20 byte sha1
  634. :raise BadObject:
  635. :return: OInfo instance, with 20 byte sha"""
  636. return self._object(sha, False)
  637. def stream(self, sha):
  638. """Retrieve an object stream along with its information as identified by the given sha
  639. :param sha: 20 byte sha1
  640. :raise BadObject:
  641. :return: OStream instance, with 20 byte sha"""
  642. return self._object(sha, True)
  643. def info_at_index(self, index):
  644. """As ``info``, but uses a PackIndexFile compatible index to refer to the object"""
  645. return self._object(None, False, index)
  646. def stream_at_index(self, index):
  647. """As ``stream``, but uses a PackIndexFile compatible index to refer to the
  648. object"""
  649. return self._object(None, True, index)
  650. #} END Read-Database like Interface
  651. #{ Interface
  652. def pack(self):
  653. """:return: the underlying pack file instance"""
  654. return self._pack
  655. def index(self):
  656. """:return: the underlying pack index file instance"""
  657. return self._index
  658. def is_valid_stream(self, sha, use_crc=False):
  659. """
  660. Verify that the stream at the given sha is valid.
  661. :param use_crc: if True, the index' crc is run over the compressed stream of
  662. the object, which is much faster than checking the sha1. It is also
  663. more prone to unnoticed corruption or manipulation.
  664. :param sha: 20 byte sha1 of the object whose stream to verify
  665. whether the compressed stream of the object is valid. If it is
  666. a delta, this only verifies that the delta's data is valid, not the
  667. data of the actual undeltified object, as it depends on more than
  668. just this stream.
  669. If False, the object will be decompressed and the sha generated. It must
  670. match the given sha
  671. :return: True if the stream is valid
  672. :raise UnsupportedOperation: If the index is version 1 only
  673. :raise BadObject: sha was not found"""
  674. if use_crc:
  675. if self._index.version() < 2:
  676. raise UnsupportedOperation("Version 1 indices do not contain crc's, verify by sha instead")
  677. # END handle index version
  678. index = self._sha_to_index(sha)
  679. offset = self._index.offset(index)
  680. next_offset = self._offset_map[offset]
  681. crc_value = self._index.crc(index)
  682. # create the current crc value, on the compressed object data
  683. # Read it in chunks, without copying the data
  684. crc_update = zlib.crc32
  685. pack_data = self._pack.data()
  686. cur_pos = offset
  687. this_crc_value = 0
  688. while cur_pos < next_offset:
  689. rbound = min(cur_pos + chunk_size, next_offset)
  690. size = rbound - cur_pos
  691. this_crc_value = crc_update(buffer(pack_data, cur_pos, size), this_crc_value)
  692. cur_pos += size
  693. # END window size loop
  694. # crc returns signed 32 bit numbers, the AND op forces it into unsigned
  695. # mode ... wow, sneaky, from dulwich.
  696. return (this_crc_value & 0xffffffff) == crc_value
  697. else:
  698. shawriter = Sha1Writer()
  699. stream = self._object(sha, as_stream=True)
  700. # write a loose object, which is the basis for the sha
  701. write_object(stream.type, stream.size, stream.read, shawriter.write)
  702. assert shawriter.sha(as_hex=False) == sha
  703. return shawriter.sha(as_hex=False) == sha
  704. # END handle crc/sha verification
  705. return True
  706. def info_iter(self):
  707. """
  708. :return: Iterator over all objects in this pack. The iterator yields
  709. OInfo instances"""
  710. return self._iter_objects(as_stream=False)
  711. def stream_iter(self):
  712. """
  713. :return: iterator over all objects in this pack. The iterator yields
  714. OStream instances"""
  715. return self._iter_objects(as_stream=True)
  716. def collect_streams_at_offset(self, offset):
  717. """
  718. As the version in the PackFile, but can resolve REF deltas within this pack
  719. For more info, see ``collect_streams``
  720. :param offset: offset into the pack file at which the object can be found"""
  721. streams = self._pack.collect_streams(offset)
  722. # try to resolve the last one if needed. It is assumed to be either
  723. # a REF delta, or a base object, as OFFSET deltas are resolved by the pack
  724. if streams[-1].type_id == REF_DELTA:
  725. stream = streams[-1]
  726. while stream.type_id in delta_types:
  727. if stream.type_id == REF_DELTA:
  728. sindex = self._index.sha_to_index(to_bytes(stream.delta_info))
  729. if sindex is None:
  730. break
  731. stream = self._pack.stream(self._index.offset(sindex))
  732. streams.append(stream)
  733. else:
  734. # must be another OFS DELTA - this could happen if a REF
  735. # delta we resolve previously points to an OFS delta. Who
  736. # would do that ;) ? We can handle it though
  737. stream = self._pack.stream(stream.delta_info)
  738. streams.append(stream)
  739. # END handle ref delta
  740. # END resolve ref streams
  741. # END resolve streams
  742. return streams
  743. def collect_streams(self, sha):
  744. """
  745. As ``PackFile.collect_streams``, but takes a sha instead of an offset.
  746. Additionally, ref_delta streams will be resolved within this pack.
  747. If this is not possible, the stream will be left alone, hence it is adivsed
  748. to check for unresolved ref-deltas and resolve them before attempting to
  749. construct a delta stream.
  750. :param sha: 20 byte sha1 specifying the object whose related streams you want to collect
  751. :return: list of streams, first being the actual object delta, the last being
  752. a possibly unresolved base object.
  753. :raise BadObject:"""
  754. return self.collect_streams_at_offset(self._index.offset(self._sha_to_index(sha)))
  755. @classmethod
  756. def write_pack(cls, object_iter, pack_write, index_write=None,
  757. object_count=None, zlib_compression=zlib.Z_BEST_SPEED):
  758. """
  759. Create a new pack by putting all objects obtained by the object_iterator
  760. into a pack which is written using the pack_write method.
  761. The respective index is produced as well if index_write is not Non.
  762. :param object_iter: iterator yielding odb output objects
  763. :param pack_write: function to receive strings to write into the pack stream
  764. :param indx_write: if not None, the function writes the index file corresponding
  765. to the pack.
  766. :param object_count: if you can provide the amount of objects in your iteration,
  767. this would be the place to put it. Otherwise we have to pre-iterate and store
  768. all items into a list to get the number, which uses more memory than necessary.
  769. :param zlib_compression: the zlib compression level to use
  770. :return: tuple(pack_sha, index_binsha) binary sha over all the contents of the pack
  771. and over all contents of the index. If index_write was None, index_binsha will be None
  772. **Note:** The destination of the write functions is up to the user. It could
  773. be a socket, or a file for instance
  774. **Note:** writes only undeltified objects"""
  775. objs = object_iter
  776. if not object_count:
  777. if not isinstance(object_iter, (tuple, list)):
  778. objs = list(object_iter)
  779. # END handle list type
  780. object_count = len(objs)
  781. # END handle object
  782. pack_writer = FlexibleSha1Writer(pack_write)
  783. pwrite = pack_writer.write
  784. ofs = 0 # current offset into the pack file
  785. index = None
  786. wants_index = index_write is not None
  787. # write header
  788. pwrite(pack('>LLL', PackFile.pack_signature, PackFile.pack_version_default, object_count))
  789. ofs += 12
  790. if wants_index:
  791. index = IndexWriter()
  792. # END handle index header
  793. actual_count = 0
  794. for obj in objs:
  795. actual_count += 1
  796. crc = 0
  797. # object header
  798. hdr = create_pack_object_header(obj.type_id, obj.size)
  799. if index_write:
  800. crc = crc32(hdr)
  801. else:
  802. crc = None
  803. # END handle crc
  804. pwrite(hdr)
  805. # data stream
  806. zstream = zlib.compressobj(zlib_compression)
  807. ostream = obj.stream
  808. br, bw, crc = write_stream_to_pack(ostream.read, pwrite, zstream, base_crc=crc)
  809. assert(br == obj.size)
  810. if wants_index:
  811. index.append(obj.binsha, crc, ofs)
  812. # END handle index
  813. ofs += len(hdr) + bw
  814. if actual_count == object_count:
  815. break
  816. # END abort once we are done
  817. # END for each object
  818. if actual_count != object_count:
  819. raise ValueError(
  820. "Expected to write %i objects into pack, but received only %i from iterators" % (object_count, actual_count))
  821. # END count assertion
  822. # write footer
  823. pack_sha = pack_writer.sha(as_hex=False)
  824. assert len(pack_sha) == 20
  825. pack_write(pack_sha)
  826. ofs += len(pack_sha) # just for completeness ;)
  827. index_sha = None
  828. if wants_index:
  829. index_sha = index.write(pack_sha, index_write)
  830. # END handle index
  831. return pack_sha, index_sha
  832. @classmethod
  833. def create(cls, object_iter, base_dir, object_count=None, zlib_compression=zlib.Z_BEST_SPEED):
  834. """Create a new on-disk entity comprised of a properly named pack file and a properly named
  835. and corresponding index file. The pack contains all OStream objects contained in object iter.
  836. :param base_dir: directory which is to contain the files
  837. :return: PackEntity instance initialized with the new pack
  838. **Note:** for more information on the other parameters see the write_pack method"""
  839. pack_fd, pack_path = tempfile.mkstemp('', 'pack', base_dir)
  840. index_fd, index_path = tempfile.mkstemp('', 'index', base_dir)
  841. pack_write = lambda d: os.write(pack_fd, d)
  842. index_write = lambda d: os.write(index_fd, d)
  843. pack_binsha, index_binsha = cls.write_pack(object_iter, pack_write, index_write, object_count, zlib_compression)
  844. os.close(pack_fd)
  845. os.close(index_fd)
  846. fmt = "pack-%s.%s"
  847. new_pack_path = os.path.join(base_dir, fmt % (bin_to_hex(pack_binsha), 'pack'))
  848. new_index_path = os.path.join(base_dir, fmt % (bin_to_hex(pack_binsha), 'idx'))
  849. os.rename(pack_path, new_pack_path)
  850. os.rename(index_path, new_index_path)
  851. return cls(new_pack_path)
  852. #} END interface