clang/utils/token-delta.py - llvm-project - Git at Google

 #!/usr/bin/env python

 from __future__ import absolute_import, division, print_function
 import os
 import re
 import subprocess
 import sys
 import tempfile

 ###


 class DeltaAlgorithm(object):
     def __init__(self):
         self.cache = set()

     def test(self, changes):
         abstract

     ###

     def getTestResult(self, changes):
         # There is no reason to cache successful tests because we will
         # always reduce the changeset when we see one.

         changeset = frozenset(changes)
         if changeset in self.cache:
             return False
         elif not self.test(changes):
             self.cache.add(changeset)
             return False
         else:
             return True

     def run(self, changes, force=False):
         # Make sure the initial test passes, if not then (a) either
         # the user doesn't expect monotonicity, and we may end up
         # doing O(N^2) tests, or (b) the test is wrong. Avoid the
         # O(N^2) case unless user requests it.
         if not force:
             if not self.getTestResult(changes):
                 raise ValueError("Initial test passed to delta fails.")

         # Check empty set first to quickly find poor test functions.
         if self.getTestResult(set()):
             return set()
         else:
             return self.delta(changes, self.split(changes))

     def split(self, S):
         """split(set) -> [sets]

         Partition a set into one or two pieces.
         """

         # There are many ways to split, we could do a better job with more
         # context information (but then the API becomes grosser).
         L = list(S)
         mid = len(L) // 2
         if mid == 0:
             return (L,)
         else:
             return L[:mid], L[mid:]

     def delta(self, c, sets):
         # assert(reduce(set.union, sets, set()) == c)

         # If there is nothing left we can remove, we are done.
         if len(sets) <= 1:
             return c

         # Look for a passing subset.
         res = self.search(c, sets)
         if res is not None:
             return res

         # Otherwise, partition sets if possible; if not we are done.
         refined = sum(map(list, map(self.split, sets)), [])
         if len(refined) == len(sets):
             return c

         return self.delta(c, refined)

     def search(self, c, sets):
         for i, S in enumerate(sets):
             # If test passes on this subset alone, recurse.
             if self.getTestResult(S):
                 return self.delta(S, self.split(S))

             # Otherwise if we have more than two sets, see if test
             # pases without this subset.
             if len(sets) > 2:
                 complement = sum(sets[:i] + sets[i + 1 :], [])
                 if self.getTestResult(complement):
                     return self.delta(complement, sets[:i] + sets[i + 1 :])


 ###


 class Token(object):
     def __init__(self, type, data, flags, file, line, column):
         self.type = type
         self.data = data
         self.flags = flags
         self.file = file
         self.line = line
         self.column = column


 kTokenRE = re.compile(
     r"""([a-z_]+) '(.*)'\t(.*)\tLoc=<(.*):(.*):(.*)>""", re.DOTALL | re.MULTILINE
 )


 def getTokens(path):
     p = subprocess.Popen(
         ["clang", "-dump-raw-tokens", path],
         stdin=subprocess.PIPE,
         stdout=subprocess.PIPE,
         stderr=subprocess.PIPE,
     )
     out, err = p.communicate()

     tokens = []
     collect = None
     for ln in err.split("\n"):
         # Silly programmers refuse to print in simple machine readable
         # formats. Whatever.
         if collect is None:
             collect = ln
         else:
             collect = collect + "\n" + ln
         if "Loc=<" in ln and ln.endswith(">"):
             ln, collect = collect, None
             tokens.append(Token(*kTokenRE.match(ln).groups()))

     return tokens


 ###


 class TMBDDelta(DeltaAlgorithm):
     def __init__(self, testProgram, tokenLists, log):
         def patchName(name, suffix):
             base, ext = os.path.splitext(name)
             return base + "." + suffix + ext

         super(TMBDDelta, self).__init__()
         self.testProgram = testProgram
         self.tokenLists = tokenLists
         self.tempFiles = [patchName(f, "tmp") for f, _ in self.tokenLists]
         self.targetFiles = [patchName(f, "ok") for f, _ in self.tokenLists]
         self.log = log
         self.numTests = 0

     def writeFiles(self, changes, fileNames):
         assert len(fileNames) == len(self.tokenLists)
         byFile = [[] for i in self.tokenLists]
         for i, j in changes:
             byFile[i].append(j)

         for i, (file, tokens) in enumerate(self.tokenLists):
             f = open(fileNames[i], "w")
             for j in byFile[i]:
                 f.write(tokens[j])
             f.close()

         return byFile

     def test(self, changes):
         self.numTests += 1

         byFile = self.writeFiles(changes, self.tempFiles)

         if self.log:
             print("TEST - ", end=" ", file=sys.stderr)
             if self.log > 1:
                 for i, (file, _) in enumerate(self.tokenLists):
                     indices = byFile[i]
                     if i:
                         sys.stderr.write("\n      ")
                     sys.stderr.write("%s:%d tokens: [" % (file, len(byFile[i])))
                     prev = None
                     for j in byFile[i]:
                         if prev is None or j != prev + 1:
                             if prev:
                                 sys.stderr.write("%d][" % prev)
                             sys.stderr.write(str(j))
                             sys.stderr.write(":")
                         prev = j
                     if byFile[i]:
                         sys.stderr.write(str(byFile[i][-1]))
                     sys.stderr.write("] ")
             else:
                 print(
                     ", ".join(
                         [
                             "%s:%d tokens" % (file, len(byFile[i]))
                             for i, (file, _) in enumerate(self.tokenLists)
                         ]
                     ),
                     end=" ",
                     file=sys.stderr,
                 )

         p = subprocess.Popen([self.testProgram] + self.tempFiles)
         res = p.wait() == 0

         if res:
             self.writeFiles(changes, self.targetFiles)

         if self.log:
             print("=> %s" % res, file=sys.stderr)
         else:
             if res:
                 print("\nSUCCESS (%d tokens)" % len(changes))
             else:
                 sys.stderr.write(".")

         return res

     def run(self):
         res = super(TMBDDelta, self).run(
             [
                 (i, j)
                 for i, (file, tokens) in enumerate(self.tokenLists)
                 for j in range(len(tokens))
             ]
         )
         self.writeFiles(res, self.targetFiles)
         if not self.log:
             print(file=sys.stderr)
         return res


 def tokenBasedMultiDelta(program, files, log):
     # Read in the lists of tokens.
     tokenLists = [(file, [t.data for t in getTokens(file)]) for file in files]

     numTokens = sum([len(tokens) for _, tokens in tokenLists])
     print("Delta on %s with %d tokens." % (", ".join(files), numTokens))

     tbmd = TMBDDelta(program, tokenLists, log)

     res = tbmd.run()

     print(
         "Finished %s with %d tokens (in %d tests)."
         % (", ".join(tbmd.targetFiles), len(res), tbmd.numTests)
     )


 def main():
     from optparse import OptionParser, OptionGroup

     parser = OptionParser("%prog <test program> {files+}")
     parser.add_option(
         "",
         "--debug",
         dest="debugLevel",
         help="set debug level [default %default]",
         action="store",
         type=int,
         default=0,
     )
     (opts, args) = parser.parse_args()

     if len(args) <= 1:
         parser.error("Invalid number of arguments.")

     program, files = args[0], args[1:]

     md = tokenBasedMultiDelta(program, files, log=opts.debugLevel)


 if __name__ == "__main__":
     try:
         main()
     except KeyboardInterrupt:
         print("Interrupted.", file=sys.stderr)
         os._exit(1)  # Avoid freeing our giant cache.
	#!/usr/bin/env python

	from __future__ import absolute_import, division, print_function
	import os
	import re
	import subprocess
	import sys
	import tempfile

	###


	class DeltaAlgorithm(object):
	def __init__(self):
	self.cache = set()

	def test(self, changes):
	abstract

	###

	def getTestResult(self, changes):
	# There is no reason to cache successful tests because we will
	# always reduce the changeset when we see one.

	changeset = frozenset(changes)
	if changeset in self.cache:
	return False
	elif not self.test(changes):
	self.cache.add(changeset)
	return False
	else:
	return True

	def run(self, changes, force=False):
	# Make sure the initial test passes, if not then (a) either
	# the user doesn't expect monotonicity, and we may end up
	# doing O(N^2) tests, or (b) the test is wrong. Avoid the
	# O(N^2) case unless user requests it.
	if not force:
	if not self.getTestResult(changes):
	raise ValueError("Initial test passed to delta fails.")

	# Check empty set first to quickly find poor test functions.
	if self.getTestResult(set()):
	return set()
	else:
	return self.delta(changes, self.split(changes))

	def split(self, S):
	"""split(set) -> [sets]

	Partition a set into one or two pieces.
	"""

	# There are many ways to split, we could do a better job with more
	# context information (but then the API becomes grosser).
	L = list(S)
	mid = len(L) // 2
	if mid == 0:
	return (L,)
	else:
	return L[:mid], L[mid:]

	def delta(self, c, sets):
	# assert(reduce(set.union, sets, set()) == c)

	# If there is nothing left we can remove, we are done.
	if len(sets) <= 1:
	return c

	# Look for a passing subset.
	res = self.search(c, sets)
	if res is not None:
	return res

	# Otherwise, partition sets if possible; if not we are done.
	refined = sum(map(list, map(self.split, sets)), [])
	if len(refined) == len(sets):
	return c

	return self.delta(c, refined)

	def search(self, c, sets):
	for i, S in enumerate(sets):
	# If test passes on this subset alone, recurse.
	if self.getTestResult(S):
	return self.delta(S, self.split(S))

	# Otherwise if we have more than two sets, see if test
	# pases without this subset.
	if len(sets) > 2:
	complement = sum(sets[:i] + sets[i + 1 :], [])
	if self.getTestResult(complement):
	return self.delta(complement, sets[:i] + sets[i + 1 :])


	###


	class Token(object):
	def __init__(self, type, data, flags, file, line, column):
	self.type = type
	self.data = data
	self.flags = flags
	self.file = file
	self.line = line
	self.column = column


	kTokenRE = re.compile(
	r"""([a-z_]+) '(.)'\t(.)\tLoc=<(.):(.):(.*)>""", re.DOTALL \| re.MULTILINE
	)


	def getTokens(path):
	p = subprocess.Popen(
	["clang", "-dump-raw-tokens", path],
	stdin=subprocess.PIPE,
	stdout=subprocess.PIPE,
	stderr=subprocess.PIPE,
	)
	out, err = p.communicate()

	tokens = []
	collect = None
	for ln in err.split("\n"):
	# Silly programmers refuse to print in simple machine readable
	# formats. Whatever.
	if collect is None:
	collect = ln
	else:
	collect = collect + "\n" + ln
	if "Loc=<" in ln and ln.endswith(">"):
	ln, collect = collect, None
	tokens.append(Token(*kTokenRE.match(ln).groups()))

	return tokens


	###


	class TMBDDelta(DeltaAlgorithm):
	def __init__(self, testProgram, tokenLists, log):
	def patchName(name, suffix):
	base, ext = os.path.splitext(name)
	return base + "." + suffix + ext

	super(TMBDDelta, self).__init__()
	self.testProgram = testProgram
	self.tokenLists = tokenLists
	self.tempFiles = [patchName(f, "tmp") for f, _ in self.tokenLists]
	self.targetFiles = [patchName(f, "ok") for f, _ in self.tokenLists]
	self.log = log
	self.numTests = 0

	def writeFiles(self, changes, fileNames):
	assert len(fileNames) == len(self.tokenLists)
	byFile = [[] for i in self.tokenLists]
	for i, j in changes:
	byFile[i].append(j)

	for i, (file, tokens) in enumerate(self.tokenLists):
	f = open(fileNames[i], "w")
	for j in byFile[i]:
	f.write(tokens[j])
	f.close()

	return byFile

	def test(self, changes):
	self.numTests += 1

	byFile = self.writeFiles(changes, self.tempFiles)

	if self.log:
	print("TEST - ", end=" ", file=sys.stderr)
	if self.log > 1:
	for i, (file, _) in enumerate(self.tokenLists):
	indices = byFile[i]
	if i:
	sys.stderr.write("\n ")
	sys.stderr.write("%s:%d tokens: [" % (file, len(byFile[i])))
	prev = None
	for j in byFile[i]:
	if prev is None or j != prev + 1:
	if prev:
	sys.stderr.write("%d][" % prev)
	sys.stderr.write(str(j))
	sys.stderr.write(":")
	prev = j
	if byFile[i]:
	sys.stderr.write(str(byFile[i][-1]))
	sys.stderr.write("] ")
	else:
	print(
	", ".join(
	[
	"%s:%d tokens" % (file, len(byFile[i]))
	for i, (file, _) in enumerate(self.tokenLists)
	]
	),
	end=" ",
	file=sys.stderr,
	)

	p = subprocess.Popen([self.testProgram] + self.tempFiles)
	res = p.wait() == 0

	if res:
	self.writeFiles(changes, self.targetFiles)

	if self.log:
	print("=> %s" % res, file=sys.stderr)
	else:
	if res:
	print("\nSUCCESS (%d tokens)" % len(changes))
	else:
	sys.stderr.write(".")

	return res

	def run(self):
	res = super(TMBDDelta, self).run(
	[
	(i, j)
	for i, (file, tokens) in enumerate(self.tokenLists)
	for j in range(len(tokens))
	]
	)
	self.writeFiles(res, self.targetFiles)
	if not self.log:
	print(file=sys.stderr)
	return res


	def tokenBasedMultiDelta(program, files, log):
	# Read in the lists of tokens.
	tokenLists = [(file, [t.data for t in getTokens(file)]) for file in files]

	numTokens = sum([len(tokens) for _, tokens in tokenLists])
	print("Delta on %s with %d tokens." % (", ".join(files), numTokens))

	tbmd = TMBDDelta(program, tokenLists, log)

	res = tbmd.run()

	print(
	"Finished %s with %d tokens (in %d tests)."
	% (", ".join(tbmd.targetFiles), len(res), tbmd.numTests)
	)


	def main():
	from optparse import OptionParser, OptionGroup

	parser = OptionParser("%prog <test program> {files+}")
	parser.add_option(
	"",
	"--debug",
	dest="debugLevel",
	help="set debug level [default %default]",
	action="store",
	type=int,
	default=0,
	)
	(opts, args) = parser.parse_args()

	if len(args) <= 1:
	parser.error("Invalid number of arguments.")

	program, files = args[0], args[1:]

	md = tokenBasedMultiDelta(program, files, log=opts.debugLevel)


	if __name__ == "__main__":
	try:
	main()
	except KeyboardInterrupt:
	print("Interrupted.", file=sys.stderr)
	os._exit(1) # Avoid freeing our giant cache.