12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075 |
- #
- # QAPI helper library
- #
- # Copyright IBM, Corp. 2011
- # Copyright (c) 2013-2016 Red Hat Inc.
- #
- # Authors:
- # Anthony Liguori <aliguori@us.ibm.com>
- # Markus Armbruster <armbru@redhat.com>
- #
- # This work is licensed under the terms of the GNU GPL, version 2.
- # See the COPYING file in the top-level directory.
- import re
- from ordereddict import OrderedDict
- import errno
- import getopt
- import os
- import sys
- import string
- builtin_types = {
- 'str': 'QTYPE_QSTRING',
- 'int': 'QTYPE_QINT',
- 'number': 'QTYPE_QFLOAT',
- 'bool': 'QTYPE_QBOOL',
- 'int8': 'QTYPE_QINT',
- 'int16': 'QTYPE_QINT',
- 'int32': 'QTYPE_QINT',
- 'int64': 'QTYPE_QINT',
- 'uint8': 'QTYPE_QINT',
- 'uint16': 'QTYPE_QINT',
- 'uint32': 'QTYPE_QINT',
- 'uint64': 'QTYPE_QINT',
- 'size': 'QTYPE_QINT',
- 'any': None, # any QType possible, actually
- 'QType': 'QTYPE_QSTRING',
- }
- # Are documentation comments required?
- doc_required = False
- # Whitelist of commands allowed to return a non-dictionary
- returns_whitelist = []
- # Whitelist of entities allowed to violate case conventions
- name_case_whitelist = []
- enum_types = []
- struct_types = []
- union_types = []
- events = []
- all_names = {}
- #
- # Parsing the schema into expressions
- #
- def error_path(parent):
- res = ''
- while parent:
- res = ('In file included from %s:%d:\n' % (parent['file'],
- parent['line'])) + res
- parent = parent['parent']
- return res
- class QAPIError(Exception):
- def __init__(self, fname, line, col, incl_info, msg):
- Exception.__init__(self)
- self.fname = fname
- self.line = line
- self.col = col
- self.info = incl_info
- self.msg = msg
- def __str__(self):
- loc = '%s:%d' % (self.fname, self.line)
- if self.col is not None:
- loc += ':%s' % self.col
- return error_path(self.info) + '%s: %s' % (loc, self.msg)
- class QAPIParseError(QAPIError):
- def __init__(self, parser, msg):
- col = 1
- for ch in parser.src[parser.line_pos:parser.pos]:
- if ch == '\t':
- col = (col + 7) % 8 + 1
- else:
- col += 1
- QAPIError.__init__(self, parser.fname, parser.line, col,
- parser.incl_info, msg)
- class QAPISemError(QAPIError):
- def __init__(self, info, msg):
- QAPIError.__init__(self, info['file'], info['line'], None,
- info['parent'], msg)
- class QAPIDoc(object):
- class Section(object):
- def __init__(self, name=None):
- # optional section name (argument/member or section name)
- self.name = name
- # the list of lines for this section
- self.content = []
- self.optional = False
- def append(self, line):
- self.content.append(line)
- def __repr__(self):
- return '\n'.join(self.content).strip()
- class ArgSection(Section):
- def __init__(self, name):
- QAPIDoc.Section.__init__(self, name)
- self.member = None
- def connect(self, member):
- self.member = member
- def __init__(self, parser, info):
- # self.parser is used to report errors with QAPIParseError. The
- # resulting error position depends on the state of the parser.
- # It happens to be the beginning of the comment. More or less
- # servicable, but action at a distance.
- self.parser = parser
- self.info = info
- self.symbol = None
- self.body = QAPIDoc.Section()
- # dict mapping parameter name to ArgSection
- self.args = OrderedDict()
- # a list of Section
- self.sections = []
- # the current section
- self.section = self.body
- # associated expression (to be set by expression parser)
- self.expr = None
- def has_section(self, name):
- """Return True if we have a section with this name."""
- for i in self.sections:
- if i.name == name:
- return True
- return False
- def append(self, line):
- """Parse a comment line and add it to the documentation."""
- line = line[1:]
- if not line:
- self._append_freeform(line)
- return
- if line[0] != ' ':
- raise QAPIParseError(self.parser, "Missing space after #")
- line = line[1:]
- # FIXME not nice: things like '# @foo:' and '# @foo: ' aren't
- # recognized, and get silently treated as ordinary text
- if self.symbol:
- self._append_symbol_line(line)
- elif not self.body.content and line.startswith('@'):
- if not line.endswith(':'):
- raise QAPIParseError(self.parser, "Line should end with :")
- self.symbol = line[1:-1]
- # FIXME invalid names other than the empty string aren't flagged
- if not self.symbol:
- raise QAPIParseError(self.parser, "Invalid name")
- else:
- self._append_freeform(line)
- def _append_symbol_line(self, line):
- name = line.split(' ', 1)[0]
- if name.startswith('@') and name.endswith(':'):
- line = line[len(name)+1:]
- self._start_args_section(name[1:-1])
- elif name in ('Returns:', 'Since:',
- # those are often singular or plural
- 'Note:', 'Notes:',
- 'Example:', 'Examples:',
- 'TODO:'):
- line = line[len(name)+1:]
- self._start_section(name[:-1])
- self._append_freeform(line)
- def _start_args_section(self, name):
- # FIXME invalid names other than the empty string aren't flagged
- if not name:
- raise QAPIParseError(self.parser, "Invalid parameter name")
- if name in self.args:
- raise QAPIParseError(self.parser,
- "'%s' parameter name duplicated" % name)
- if self.sections:
- raise QAPIParseError(self.parser,
- "'@%s:' can't follow '%s' section"
- % (name, self.sections[0].name))
- self.section = QAPIDoc.ArgSection(name)
- self.args[name] = self.section
- def _start_section(self, name=''):
- if name in ('Returns', 'Since') and self.has_section(name):
- raise QAPIParseError(self.parser,
- "Duplicated '%s' section" % name)
- self.section = QAPIDoc.Section(name)
- self.sections.append(self.section)
- def _append_freeform(self, line):
- in_arg = isinstance(self.section, QAPIDoc.ArgSection)
- if (in_arg and self.section.content
- and not self.section.content[-1]
- and line and not line[0].isspace()):
- self._start_section()
- if (in_arg or not self.section.name
- or not self.section.name.startswith('Example')):
- line = line.strip()
- match = re.match(r'(@\S+:)', line)
- if match:
- raise QAPIParseError(self.parser,
- "'%s' not allowed in free-form documentation"
- % match.group(1))
- # TODO Drop this once the dust has settled
- if (isinstance(self.section, QAPIDoc.ArgSection)
- and '#optional' in line):
- raise QAPISemError(self.info, "Please drop the #optional tag")
- self.section.append(line)
- def connect_member(self, member):
- if member.name not in self.args:
- # Undocumented TODO outlaw
- self.args[member.name] = QAPIDoc.ArgSection(member.name)
- self.args[member.name].connect(member)
- class QAPISchemaParser(object):
- def __init__(self, fp, previously_included=[], incl_info=None):
- abs_fname = os.path.abspath(fp.name)
- fname = fp.name
- self.fname = fname
- previously_included.append(abs_fname)
- self.incl_info = incl_info
- self.src = fp.read()
- if self.src == '' or self.src[-1] != '\n':
- self.src += '\n'
- self.cursor = 0
- self.line = 1
- self.line_pos = 0
- self.exprs = []
- self.docs = []
- self.cur_doc = None
- self.accept()
- while self.tok is not None:
- info = {'file': fname, 'line': self.line,
- 'parent': self.incl_info}
- if self.tok == '#':
- self.reject_expr_doc()
- self.cur_doc = self.get_doc(info)
- self.docs.append(self.cur_doc)
- continue
- expr = self.get_expr(False)
- if 'include' in expr:
- self.reject_expr_doc()
- if len(expr) != 1:
- raise QAPISemError(info, "Invalid 'include' directive")
- include = expr['include']
- if not isinstance(include, str):
- raise QAPISemError(info,
- "Value of 'include' must be a string")
- self._include(include, info, os.path.dirname(abs_fname),
- previously_included)
- elif "pragma" in expr:
- self.reject_expr_doc()
- if len(expr) != 1:
- raise QAPISemError(info, "Invalid 'pragma' directive")
- pragma = expr['pragma']
- if not isinstance(pragma, dict):
- raise QAPISemError(
- info, "Value of 'pragma' must be a dictionary")
- for name, value in pragma.iteritems():
- self._pragma(name, value, info)
- else:
- expr_elem = {'expr': expr,
- 'info': info}
- if self.cur_doc:
- if not self.cur_doc.symbol:
- raise QAPISemError(
- self.cur_doc.info,
- "Expression documentation required")
- self.cur_doc.expr = expr
- expr_elem['doc'] = self.cur_doc
- self.exprs.append(expr_elem)
- self.cur_doc = None
- self.reject_expr_doc()
- def reject_expr_doc(self):
- if self.cur_doc and self.cur_doc.symbol:
- raise QAPISemError(
- self.cur_doc.info,
- "Documentation for '%s' is not followed by the definition"
- % self.cur_doc.symbol)
- def _include(self, include, info, base_dir, previously_included):
- incl_abs_fname = os.path.join(base_dir, include)
- # catch inclusion cycle
- inf = info
- while inf:
- if incl_abs_fname == os.path.abspath(inf['file']):
- raise QAPISemError(info, "Inclusion loop for %s" % include)
- inf = inf['parent']
- # skip multiple include of the same file
- if incl_abs_fname in previously_included:
- return
- try:
- fobj = open(incl_abs_fname, 'r')
- except IOError as e:
- raise QAPISemError(info, '%s: %s' % (e.strerror, include))
- exprs_include = QAPISchemaParser(fobj, previously_included, info)
- self.exprs.extend(exprs_include.exprs)
- self.docs.extend(exprs_include.docs)
- def _pragma(self, name, value, info):
- global doc_required, returns_whitelist, name_case_whitelist
- if name == 'doc-required':
- if not isinstance(value, bool):
- raise QAPISemError(info,
- "Pragma 'doc-required' must be boolean")
- doc_required = value
- elif name == 'returns-whitelist':
- if (not isinstance(value, list)
- or any([not isinstance(elt, str) for elt in value])):
- raise QAPISemError(info,
- "Pragma returns-whitelist must be"
- " a list of strings")
- returns_whitelist = value
- elif name == 'name-case-whitelist':
- if (not isinstance(value, list)
- or any([not isinstance(elt, str) for elt in value])):
- raise QAPISemError(info,
- "Pragma name-case-whitelist must be"
- " a list of strings")
- name_case_whitelist = value
- else:
- raise QAPISemError(info, "Unknown pragma '%s'" % name)
- def accept(self, skip_comment=True):
- while True:
- self.tok = self.src[self.cursor]
- self.pos = self.cursor
- self.cursor += 1
- self.val = None
- if self.tok == '#':
- if self.src[self.cursor] == '#':
- # Start of doc comment
- skip_comment = False
- self.cursor = self.src.find('\n', self.cursor)
- if not skip_comment:
- self.val = self.src[self.pos:self.cursor]
- return
- elif self.tok in '{}:,[]':
- return
- elif self.tok == "'":
- string = ''
- esc = False
- while True:
- ch = self.src[self.cursor]
- self.cursor += 1
- if ch == '\n':
- raise QAPIParseError(self, 'Missing terminating "\'"')
- if esc:
- if ch == 'b':
- string += '\b'
- elif ch == 'f':
- string += '\f'
- elif ch == 'n':
- string += '\n'
- elif ch == 'r':
- string += '\r'
- elif ch == 't':
- string += '\t'
- elif ch == 'u':
- value = 0
- for _ in range(0, 4):
- ch = self.src[self.cursor]
- self.cursor += 1
- if ch not in '0123456789abcdefABCDEF':
- raise QAPIParseError(self,
- '\\u escape needs 4 '
- 'hex digits')
- value = (value << 4) + int(ch, 16)
- # If Python 2 and 3 didn't disagree so much on
- # how to handle Unicode, then we could allow
- # Unicode string defaults. But most of QAPI is
- # ASCII-only, so we aren't losing much for now.
- if not value or value > 0x7f:
- raise QAPIParseError(self,
- 'For now, \\u escape '
- 'only supports non-zero '
- 'values up to \\u007f')
- string += chr(value)
- elif ch in '\\/\'"':
- string += ch
- else:
- raise QAPIParseError(self,
- "Unknown escape \\%s" % ch)
- esc = False
- elif ch == '\\':
- esc = True
- elif ch == "'":
- self.val = string
- return
- else:
- string += ch
- elif self.src.startswith('true', self.pos):
- self.val = True
- self.cursor += 3
- return
- elif self.src.startswith('false', self.pos):
- self.val = False
- self.cursor += 4
- return
- elif self.src.startswith('null', self.pos):
- self.val = None
- self.cursor += 3
- return
- elif self.tok == '\n':
- if self.cursor == len(self.src):
- self.tok = None
- return
- self.line += 1
- self.line_pos = self.cursor
- elif not self.tok.isspace():
- raise QAPIParseError(self, 'Stray "%s"' % self.tok)
- def get_members(self):
- expr = OrderedDict()
- if self.tok == '}':
- self.accept()
- return expr
- if self.tok != "'":
- raise QAPIParseError(self, 'Expected string or "}"')
- while True:
- key = self.val
- self.accept()
- if self.tok != ':':
- raise QAPIParseError(self, 'Expected ":"')
- self.accept()
- if key in expr:
- raise QAPIParseError(self, 'Duplicate key "%s"' % key)
- expr[key] = self.get_expr(True)
- if self.tok == '}':
- self.accept()
- return expr
- if self.tok != ',':
- raise QAPIParseError(self, 'Expected "," or "}"')
- self.accept()
- if self.tok != "'":
- raise QAPIParseError(self, 'Expected string')
- def get_values(self):
- expr = []
- if self.tok == ']':
- self.accept()
- return expr
- if self.tok not in "{['tfn":
- raise QAPIParseError(self, 'Expected "{", "[", "]", string, '
- 'boolean or "null"')
- while True:
- expr.append(self.get_expr(True))
- if self.tok == ']':
- self.accept()
- return expr
- if self.tok != ',':
- raise QAPIParseError(self, 'Expected "," or "]"')
- self.accept()
- def get_expr(self, nested):
- if self.tok != '{' and not nested:
- raise QAPIParseError(self, 'Expected "{"')
- if self.tok == '{':
- self.accept()
- expr = self.get_members()
- elif self.tok == '[':
- self.accept()
- expr = self.get_values()
- elif self.tok in "'tfn":
- expr = self.val
- self.accept()
- else:
- raise QAPIParseError(self, 'Expected "{", "[" or string')
- return expr
- def get_doc(self, info):
- if self.val != '##':
- raise QAPIParseError(self, "Junk after '##' at start of "
- "documentation comment")
- doc = QAPIDoc(self, info)
- self.accept(False)
- while self.tok == '#':
- if self.val.startswith('##'):
- # End of doc comment
- if self.val != '##':
- raise QAPIParseError(self, "Junk after '##' at end of "
- "documentation comment")
- self.accept()
- return doc
- else:
- doc.append(self.val)
- self.accept(False)
- raise QAPIParseError(self, "Documentation comment must end with '##'")
- #
- # Semantic analysis of schema expressions
- # TODO fold into QAPISchema
- # TODO catching name collisions in generated code would be nice
- #
- def find_base_members(base):
- if isinstance(base, dict):
- return base
- base_struct_define = find_struct(base)
- if not base_struct_define:
- return None
- return base_struct_define['data']
- # Return the qtype of an alternate branch, or None on error.
- def find_alternate_member_qtype(qapi_type):
- if qapi_type in builtin_types:
- return builtin_types[qapi_type]
- elif find_struct(qapi_type):
- return 'QTYPE_QDICT'
- elif find_enum(qapi_type):
- return 'QTYPE_QSTRING'
- elif find_union(qapi_type):
- return 'QTYPE_QDICT'
- return None
- # Return the discriminator enum define if discriminator is specified as an
- # enum type, otherwise return None.
- def discriminator_find_enum_define(expr):
- base = expr.get('base')
- discriminator = expr.get('discriminator')
- if not (discriminator and base):
- return None
- base_members = find_base_members(base)
- if not base_members:
- return None
- discriminator_type = base_members.get(discriminator)
- if not discriminator_type:
- return None
- return find_enum(discriminator_type)
- # Names must be letters, numbers, -, and _. They must start with letter,
- # except for downstream extensions which must start with __RFQDN_.
- # Dots are only valid in the downstream extension prefix.
- valid_name = re.compile(r'^(__[a-zA-Z0-9.-]+_)?'
- '[a-zA-Z][a-zA-Z0-9_-]*$')
- def check_name(info, source, name, allow_optional=False,
- enum_member=False):
- global valid_name
- membername = name
- if not isinstance(name, str):
- raise QAPISemError(info, "%s requires a string name" % source)
- if name.startswith('*'):
- membername = name[1:]
- if not allow_optional:
- raise QAPISemError(info, "%s does not allow optional name '%s'"
- % (source, name))
- # Enum members can start with a digit, because the generated C
- # code always prefixes it with the enum name
- if enum_member and membername[0].isdigit():
- membername = 'D' + membername
- # Reserve the entire 'q_' namespace for c_name(), and for 'q_empty'
- # and 'q_obj_*' implicit type names.
- if not valid_name.match(membername) or \
- c_name(membername, False).startswith('q_'):
- raise QAPISemError(info, "%s uses invalid name '%s'" % (source, name))
- def add_name(name, info, meta, implicit=False):
- global all_names
- check_name(info, "'%s'" % meta, name)
- # FIXME should reject names that differ only in '_' vs. '.'
- # vs. '-', because they're liable to clash in generated C.
- if name in all_names:
- raise QAPISemError(info, "%s '%s' is already defined"
- % (all_names[name], name))
- if not implicit and (name.endswith('Kind') or name.endswith('List')):
- raise QAPISemError(info, "%s '%s' should not end in '%s'"
- % (meta, name, name[-4:]))
- all_names[name] = meta
- def add_struct(definition, info):
- global struct_types
- name = definition['struct']
- add_name(name, info, 'struct')
- struct_types.append(definition)
- def find_struct(name):
- global struct_types
- for struct in struct_types:
- if struct['struct'] == name:
- return struct
- return None
- def add_union(definition, info):
- global union_types
- name = definition['union']
- add_name(name, info, 'union')
- union_types.append(definition)
- def find_union(name):
- global union_types
- for union in union_types:
- if union['union'] == name:
- return union
- return None
- def add_enum(name, info, enum_values=None, implicit=False):
- global enum_types
- add_name(name, info, 'enum', implicit)
- enum_types.append({'enum_name': name, 'enum_values': enum_values})
- def find_enum(name):
- global enum_types
- for enum in enum_types:
- if enum['enum_name'] == name:
- return enum
- return None
- def is_enum(name):
- return find_enum(name) is not None
- def check_type(info, source, value, allow_array=False,
- allow_dict=False, allow_optional=False,
- allow_metas=[]):
- global all_names
- if value is None:
- return
- # Check if array type for value is okay
- if isinstance(value, list):
- if not allow_array:
- raise QAPISemError(info, "%s cannot be an array" % source)
- if len(value) != 1 or not isinstance(value[0], str):
- raise QAPISemError(info,
- "%s: array type must contain single type name" %
- source)
- value = value[0]
- # Check if type name for value is okay
- if isinstance(value, str):
- if value not in all_names:
- raise QAPISemError(info, "%s uses unknown type '%s'"
- % (source, value))
- if not all_names[value] in allow_metas:
- raise QAPISemError(info, "%s cannot use %s type '%s'" %
- (source, all_names[value], value))
- return
- if not allow_dict:
- raise QAPISemError(info, "%s should be a type name" % source)
- if not isinstance(value, OrderedDict):
- raise QAPISemError(info,
- "%s should be a dictionary or type name" % source)
- # value is a dictionary, check that each member is okay
- for (key, arg) in value.items():
- check_name(info, "Member of %s" % source, key,
- allow_optional=allow_optional)
- if c_name(key, False) == 'u' or c_name(key, False).startswith('has_'):
- raise QAPISemError(info, "Member of %s uses reserved name '%s'"
- % (source, key))
- # Todo: allow dictionaries to represent default values of
- # an optional argument.
- check_type(info, "Member '%s' of %s" % (key, source), arg,
- allow_array=True,
- allow_metas=['built-in', 'union', 'alternate', 'struct',
- 'enum'])
- def check_command(expr, info):
- name = expr['command']
- boxed = expr.get('boxed', False)
- args_meta = ['struct']
- if boxed:
- args_meta += ['union', 'alternate']
- check_type(info, "'data' for command '%s'" % name,
- expr.get('data'), allow_dict=not boxed, allow_optional=True,
- allow_metas=args_meta)
- returns_meta = ['union', 'struct']
- if name in returns_whitelist:
- returns_meta += ['built-in', 'alternate', 'enum']
- check_type(info, "'returns' for command '%s'" % name,
- expr.get('returns'), allow_array=True,
- allow_optional=True, allow_metas=returns_meta)
- def check_event(expr, info):
- global events
- name = expr['event']
- boxed = expr.get('boxed', False)
- meta = ['struct']
- if boxed:
- meta += ['union', 'alternate']
- events.append(name)
- check_type(info, "'data' for event '%s'" % name,
- expr.get('data'), allow_dict=not boxed, allow_optional=True,
- allow_metas=meta)
- def check_union(expr, info):
- name = expr['union']
- base = expr.get('base')
- discriminator = expr.get('discriminator')
- members = expr['data']
- # Two types of unions, determined by discriminator.
- # With no discriminator it is a simple union.
- if discriminator is None:
- enum_define = None
- allow_metas = ['built-in', 'union', 'alternate', 'struct', 'enum']
- if base is not None:
- raise QAPISemError(info, "Simple union '%s' must not have a base" %
- name)
- # Else, it's a flat union.
- else:
- # The object must have a string or dictionary 'base'.
- check_type(info, "'base' for union '%s'" % name,
- base, allow_dict=True, allow_optional=True,
- allow_metas=['struct'])
- if not base:
- raise QAPISemError(info, "Flat union '%s' must have a base"
- % name)
- base_members = find_base_members(base)
- assert base_members is not None
- # The value of member 'discriminator' must name a non-optional
- # member of the base struct.
- check_name(info, "Discriminator of flat union '%s'" % name,
- discriminator)
- discriminator_type = base_members.get(discriminator)
- if not discriminator_type:
- raise QAPISemError(info,
- "Discriminator '%s' is not a member of base "
- "struct '%s'"
- % (discriminator, base))
- enum_define = find_enum(discriminator_type)
- allow_metas = ['struct']
- # Do not allow string discriminator
- if not enum_define:
- raise QAPISemError(info,
- "Discriminator '%s' must be of enumeration "
- "type" % discriminator)
- # Check every branch; don't allow an empty union
- if len(members) == 0:
- raise QAPISemError(info, "Union '%s' cannot have empty 'data'" % name)
- for (key, value) in members.items():
- check_name(info, "Member of union '%s'" % name, key)
- # Each value must name a known type
- check_type(info, "Member '%s' of union '%s'" % (key, name),
- value, allow_array=not base, allow_metas=allow_metas)
- # If the discriminator names an enum type, then all members
- # of 'data' must also be members of the enum type.
- if enum_define:
- if key not in enum_define['enum_values']:
- raise QAPISemError(info,
- "Discriminator value '%s' is not found in "
- "enum '%s'"
- % (key, enum_define['enum_name']))
- # If discriminator is user-defined, ensure all values are covered
- if enum_define:
- for value in enum_define['enum_values']:
- if value not in members.keys():
- raise QAPISemError(info, "Union '%s' data missing '%s' branch"
- % (name, value))
- def check_alternate(expr, info):
- name = expr['alternate']
- members = expr['data']
- types_seen = {}
- # Check every branch; require at least two branches
- if len(members) < 2:
- raise QAPISemError(info,
- "Alternate '%s' should have at least two branches "
- "in 'data'" % name)
- for (key, value) in members.items():
- check_name(info, "Member of alternate '%s'" % name, key)
- # Ensure alternates have no type conflicts.
- check_type(info, "Member '%s' of alternate '%s'" % (key, name),
- value,
- allow_metas=['built-in', 'union', 'struct', 'enum'])
- qtype = find_alternate_member_qtype(value)
- if not qtype:
- raise QAPISemError(info, "Alternate '%s' member '%s' cannot use "
- "type '%s'" % (name, key, value))
- if qtype in types_seen:
- raise QAPISemError(info, "Alternate '%s' member '%s' can't "
- "be distinguished from member '%s'"
- % (name, key, types_seen[qtype]))
- types_seen[qtype] = key
- def check_enum(expr, info):
- name = expr['enum']
- members = expr.get('data')
- prefix = expr.get('prefix')
- if not isinstance(members, list):
- raise QAPISemError(info,
- "Enum '%s' requires an array for 'data'" % name)
- if prefix is not None and not isinstance(prefix, str):
- raise QAPISemError(info,
- "Enum '%s' requires a string for 'prefix'" % name)
- for member in members:
- check_name(info, "Member of enum '%s'" % name, member,
- enum_member=True)
- def check_struct(expr, info):
- name = expr['struct']
- members = expr['data']
- check_type(info, "'data' for struct '%s'" % name, members,
- allow_dict=True, allow_optional=True)
- check_type(info, "'base' for struct '%s'" % name, expr.get('base'),
- allow_metas=['struct'])
- def check_keys(expr_elem, meta, required, optional=[]):
- expr = expr_elem['expr']
- info = expr_elem['info']
- name = expr[meta]
- if not isinstance(name, str):
- raise QAPISemError(info, "'%s' key must have a string value" % meta)
- required = required + [meta]
- for (key, value) in expr.items():
- if key not in required and key not in optional:
- raise QAPISemError(info, "Unknown key '%s' in %s '%s'"
- % (key, meta, name))
- if (key == 'gen' or key == 'success-response') and value is not False:
- raise QAPISemError(info,
- "'%s' of %s '%s' should only use false value"
- % (key, meta, name))
- if key == 'boxed' and value is not True:
- raise QAPISemError(info,
- "'%s' of %s '%s' should only use true value"
- % (key, meta, name))
- for key in required:
- if key not in expr:
- raise QAPISemError(info, "Key '%s' is missing from %s '%s'"
- % (key, meta, name))
- def check_exprs(exprs):
- global all_names
- # Populate name table with names of built-in types
- for builtin in builtin_types.keys():
- all_names[builtin] = 'built-in'
- # Learn the types and check for valid expression keys
- for expr_elem in exprs:
- expr = expr_elem['expr']
- info = expr_elem['info']
- doc = expr_elem.get('doc')
- if not doc and doc_required:
- raise QAPISemError(info,
- "Expression missing documentation comment")
- if 'enum' in expr:
- name = expr['enum']
- check_keys(expr_elem, 'enum', ['data'], ['prefix'])
- add_enum(name, info, expr['data'])
- elif 'union' in expr:
- name = expr['union']
- check_keys(expr_elem, 'union', ['data'],
- ['base', 'discriminator'])
- add_union(expr, info)
- elif 'alternate' in expr:
- name = expr['alternate']
- check_keys(expr_elem, 'alternate', ['data'])
- add_name(name, info, 'alternate')
- elif 'struct' in expr:
- name = expr['struct']
- check_keys(expr_elem, 'struct', ['data'], ['base'])
- add_struct(expr, info)
- elif 'command' in expr:
- name = expr['command']
- check_keys(expr_elem, 'command', [],
- ['data', 'returns', 'gen', 'success-response', 'boxed'])
- add_name(name, info, 'command')
- elif 'event' in expr:
- name = expr['event']
- check_keys(expr_elem, 'event', [], ['data', 'boxed'])
- add_name(name, info, 'event')
- else:
- raise QAPISemError(expr_elem['info'],
- "Expression is missing metatype")
- if doc and doc.symbol != name:
- raise QAPISemError(info, "Definition of '%s' follows documentation"
- " for '%s'" % (name, doc.symbol))
- # Try again for hidden UnionKind enum
- for expr_elem in exprs:
- expr = expr_elem['expr']
- if 'union' in expr:
- if not discriminator_find_enum_define(expr):
- add_enum('%sKind' % expr['union'], expr_elem['info'],
- implicit=True)
- elif 'alternate' in expr:
- add_enum('%sKind' % expr['alternate'], expr_elem['info'],
- implicit=True)
- # Validate that exprs make sense
- for expr_elem in exprs:
- expr = expr_elem['expr']
- info = expr_elem['info']
- if 'enum' in expr:
- check_enum(expr, info)
- elif 'union' in expr:
- check_union(expr, info)
- elif 'alternate' in expr:
- check_alternate(expr, info)
- elif 'struct' in expr:
- check_struct(expr, info)
- elif 'command' in expr:
- check_command(expr, info)
- elif 'event' in expr:
- check_event(expr, info)
- else:
- assert False, 'unexpected meta type'
- return exprs
- def check_definition_doc(doc, expr, info):
- for i in ('enum', 'union', 'alternate', 'struct', 'command', 'event'):
- if i in expr:
- meta = i
- break
- if doc.has_section('Returns') and 'command' not in expr:
- raise QAPISemError(info, "'Returns:' is only valid for commands")
- if meta == 'union':
- args = expr.get('base', [])
- else:
- args = expr.get('data', [])
- if isinstance(args, str):
- return
- if isinstance(args, dict):
- args = args.keys()
- assert isinstance(args, list)
- if (meta == 'alternate'
- or (meta == 'union' and not expr.get('discriminator'))):
- args.append('type')
- doc_args = set(doc.args.keys())
- args = set([name.strip('*') for name in args])
- if not doc_args.issubset(args):
- raise QAPISemError(info, "The following documented members are not in "
- "the declaration: %s" % ', '.join(doc_args - args))
- def check_docs(docs):
- for doc in docs:
- for section in doc.args.values() + doc.sections:
- content = str(section)
- if not content or content.isspace():
- raise QAPISemError(doc.info,
- "Empty doc section '%s'" % section.name)
- if doc.expr:
- check_definition_doc(doc, doc.expr, doc.info)
- return docs
- #
- # Schema compiler frontend
- #
- class QAPISchemaEntity(object):
- def __init__(self, name, info, doc):
- assert isinstance(name, str)
- self.name = name
- # For explicitly defined entities, info points to the (explicit)
- # definition. For builtins (and their arrays), info is None.
- # For implicitly defined entities, info points to a place that
- # triggered the implicit definition (there may be more than one
- # such place).
- self.info = info
- self.doc = doc
- def c_name(self):
- return c_name(self.name)
- def check(self, schema):
- pass
- def is_implicit(self):
- return not self.info
- def visit(self, visitor):
- pass
- class QAPISchemaVisitor(object):
- def visit_begin(self, schema):
- pass
- def visit_end(self):
- pass
- def visit_needed(self, entity):
- # Default to visiting everything
- return True
- def visit_builtin_type(self, name, info, json_type):
- pass
- def visit_enum_type(self, name, info, values, prefix):
- pass
- def visit_array_type(self, name, info, element_type):
- pass
- def visit_object_type(self, name, info, base, members, variants):
- pass
- def visit_object_type_flat(self, name, info, members, variants):
- pass
- def visit_alternate_type(self, name, info, variants):
- pass
- def visit_command(self, name, info, arg_type, ret_type,
- gen, success_response, boxed):
- pass
- def visit_event(self, name, info, arg_type, boxed):
- pass
- class QAPISchemaType(QAPISchemaEntity):
- # Return the C type for common use.
- # For the types we commonly box, this is a pointer type.
- def c_type(self):
- pass
- # Return the C type to be used in a parameter list.
- def c_param_type(self):
- return self.c_type()
- # Return the C type to be used where we suppress boxing.
- def c_unboxed_type(self):
- return self.c_type()
- def json_type(self):
- pass
- def alternate_qtype(self):
- json2qtype = {
- 'string': 'QTYPE_QSTRING',
- 'number': 'QTYPE_QFLOAT',
- 'int': 'QTYPE_QINT',
- 'boolean': 'QTYPE_QBOOL',
- 'object': 'QTYPE_QDICT'
- }
- return json2qtype.get(self.json_type())
- def doc_type(self):
- if self.is_implicit():
- return None
- return self.name
- class QAPISchemaBuiltinType(QAPISchemaType):
- def __init__(self, name, json_type, c_type):
- QAPISchemaType.__init__(self, name, None, None)
- assert not c_type or isinstance(c_type, str)
- assert json_type in ('string', 'number', 'int', 'boolean', 'null',
- 'value')
- self._json_type_name = json_type
- self._c_type_name = c_type
- def c_name(self):
- return self.name
- def c_type(self):
- return self._c_type_name
- def c_param_type(self):
- if self.name == 'str':
- return 'const ' + self._c_type_name
- return self._c_type_name
- def json_type(self):
- return self._json_type_name
- def doc_type(self):
- return self.json_type()
- def visit(self, visitor):
- visitor.visit_builtin_type(self.name, self.info, self.json_type())
- class QAPISchemaEnumType(QAPISchemaType):
- def __init__(self, name, info, doc, values, prefix):
- QAPISchemaType.__init__(self, name, info, doc)
- for v in values:
- assert isinstance(v, QAPISchemaMember)
- v.set_owner(name)
- assert prefix is None or isinstance(prefix, str)
- self.values = values
- self.prefix = prefix
- def check(self, schema):
- seen = {}
- for v in self.values:
- v.check_clash(self.info, seen)
- if self.doc:
- self.doc.connect_member(v)
- def is_implicit(self):
- # See QAPISchema._make_implicit_enum_type() and ._def_predefineds()
- return self.name.endswith('Kind') or self.name == 'QType'
- def c_type(self):
- return c_name(self.name)
- def member_names(self):
- return [v.name for v in self.values]
- def json_type(self):
- return 'string'
- def visit(self, visitor):
- visitor.visit_enum_type(self.name, self.info,
- self.member_names(), self.prefix)
- class QAPISchemaArrayType(QAPISchemaType):
- def __init__(self, name, info, element_type):
- QAPISchemaType.__init__(self, name, info, None)
- assert isinstance(element_type, str)
- self._element_type_name = element_type
- self.element_type = None
- def check(self, schema):
- self.element_type = schema.lookup_type(self._element_type_name)
- assert self.element_type
- def is_implicit(self):
- return True
- def c_type(self):
- return c_name(self.name) + pointer_suffix
- def json_type(self):
- return 'array'
- def doc_type(self):
- elt_doc_type = self.element_type.doc_type()
- if not elt_doc_type:
- return None
- return 'array of ' + elt_doc_type
- def visit(self, visitor):
- visitor.visit_array_type(self.name, self.info, self.element_type)
- class QAPISchemaObjectType(QAPISchemaType):
- def __init__(self, name, info, doc, base, local_members, variants):
- # struct has local_members, optional base, and no variants
- # flat union has base, variants, and no local_members
- # simple union has local_members, variants, and no base
- QAPISchemaType.__init__(self, name, info, doc)
- assert base is None or isinstance(base, str)
- for m in local_members:
- assert isinstance(m, QAPISchemaObjectTypeMember)
- m.set_owner(name)
- if variants is not None:
- assert isinstance(variants, QAPISchemaObjectTypeVariants)
- variants.set_owner(name)
- self._base_name = base
- self.base = None
- self.local_members = local_members
- self.variants = variants
- self.members = None
- def check(self, schema):
- if self.members is False: # check for cycles
- raise QAPISemError(self.info,
- "Object %s contains itself" % self.name)
- if self.members:
- return
- self.members = False # mark as being checked
- seen = OrderedDict()
- if self._base_name:
- self.base = schema.lookup_type(self._base_name)
- assert isinstance(self.base, QAPISchemaObjectType)
- self.base.check(schema)
- self.base.check_clash(schema, self.info, seen)
- for m in self.local_members:
- m.check(schema)
- m.check_clash(self.info, seen)
- if self.doc:
- self.doc.connect_member(m)
- self.members = seen.values()
- if self.variants:
- self.variants.check(schema, seen)
- assert self.variants.tag_member in self.members
- self.variants.check_clash(schema, self.info, seen)
- # Check that the members of this type do not cause duplicate JSON members,
- # and update seen to track the members seen so far. Report any errors
- # on behalf of info, which is not necessarily self.info
- def check_clash(self, schema, info, seen):
- assert not self.variants # not implemented
- for m in self.members:
- m.check_clash(info, seen)
- def is_implicit(self):
- # See QAPISchema._make_implicit_object_type(), as well as
- # _def_predefineds()
- return self.name.startswith('q_')
- def is_empty(self):
- assert self.members is not None
- return not self.members and not self.variants
- def c_name(self):
- assert self.name != 'q_empty'
- return QAPISchemaType.c_name(self)
- def c_type(self):
- assert not self.is_implicit()
- return c_name(self.name) + pointer_suffix
- def c_unboxed_type(self):
- return c_name(self.name)
- def json_type(self):
- return 'object'
- def visit(self, visitor):
- visitor.visit_object_type(self.name, self.info,
- self.base, self.local_members, self.variants)
- visitor.visit_object_type_flat(self.name, self.info,
- self.members, self.variants)
- class QAPISchemaMember(object):
- role = 'member'
- def __init__(self, name):
- assert isinstance(name, str)
- self.name = name
- self.owner = None
- def set_owner(self, name):
- assert not self.owner
- self.owner = name
- def check_clash(self, info, seen):
- cname = c_name(self.name)
- if cname.lower() != cname and self.owner not in name_case_whitelist:
- raise QAPISemError(info,
- "%s should not use uppercase" % self.describe())
- if cname in seen:
- raise QAPISemError(info, "%s collides with %s" %
- (self.describe(), seen[cname].describe()))
- seen[cname] = self
- def _pretty_owner(self):
- owner = self.owner
- if owner.startswith('q_obj_'):
- # See QAPISchema._make_implicit_object_type() - reverse the
- # mapping there to create a nice human-readable description
- owner = owner[6:]
- if owner.endswith('-arg'):
- return '(parameter of %s)' % owner[:-4]
- elif owner.endswith('-base'):
- return '(base of %s)' % owner[:-5]
- else:
- assert owner.endswith('-wrapper')
- # Unreachable and not implemented
- assert False
- if owner.endswith('Kind'):
- # See QAPISchema._make_implicit_enum_type()
- return '(branch of %s)' % owner[:-4]
- return '(%s of %s)' % (self.role, owner)
- def describe(self):
- return "'%s' %s" % (self.name, self._pretty_owner())
- class QAPISchemaObjectTypeMember(QAPISchemaMember):
- def __init__(self, name, typ, optional):
- QAPISchemaMember.__init__(self, name)
- assert isinstance(typ, str)
- assert isinstance(optional, bool)
- self._type_name = typ
- self.type = None
- self.optional = optional
- def check(self, schema):
- assert self.owner
- self.type = schema.lookup_type(self._type_name)
- assert self.type
- class QAPISchemaObjectTypeVariants(object):
- def __init__(self, tag_name, tag_member, variants):
- # Flat unions pass tag_name but not tag_member.
- # Simple unions and alternates pass tag_member but not tag_name.
- # After check(), tag_member is always set, and tag_name remains
- # a reliable witness of being used by a flat union.
- assert bool(tag_member) != bool(tag_name)
- assert (isinstance(tag_name, str) or
- isinstance(tag_member, QAPISchemaObjectTypeMember))
- assert len(variants) > 0
- for v in variants:
- assert isinstance(v, QAPISchemaObjectTypeVariant)
- self._tag_name = tag_name
- self.tag_member = tag_member
- self.variants = variants
- def set_owner(self, name):
- for v in self.variants:
- v.set_owner(name)
- def check(self, schema, seen):
- if not self.tag_member: # flat union
- self.tag_member = seen[c_name(self._tag_name)]
- assert self._tag_name == self.tag_member.name
- assert isinstance(self.tag_member.type, QAPISchemaEnumType)
- for v in self.variants:
- v.check(schema)
- # Union names must match enum values; alternate names are
- # checked separately. Use 'seen' to tell the two apart.
- if seen:
- assert v.name in self.tag_member.type.member_names()
- assert isinstance(v.type, QAPISchemaObjectType)
- v.type.check(schema)
- def check_clash(self, schema, info, seen):
- for v in self.variants:
- # Reset seen map for each variant, since qapi names from one
- # branch do not affect another branch
- assert isinstance(v.type, QAPISchemaObjectType)
- v.type.check_clash(schema, info, dict(seen))
- class QAPISchemaObjectTypeVariant(QAPISchemaObjectTypeMember):
- role = 'branch'
- def __init__(self, name, typ):
- QAPISchemaObjectTypeMember.__init__(self, name, typ, False)
- class QAPISchemaAlternateType(QAPISchemaType):
- def __init__(self, name, info, doc, variants):
- QAPISchemaType.__init__(self, name, info, doc)
- assert isinstance(variants, QAPISchemaObjectTypeVariants)
- assert variants.tag_member
- variants.set_owner(name)
- variants.tag_member.set_owner(self.name)
- self.variants = variants
- def check(self, schema):
- self.variants.tag_member.check(schema)
- # Not calling self.variants.check_clash(), because there's nothing
- # to clash with
- self.variants.check(schema, {})
- # Alternate branch names have no relation to the tag enum values;
- # so we have to check for potential name collisions ourselves.
- seen = {}
- for v in self.variants.variants:
- v.check_clash(self.info, seen)
- if self.doc:
- self.doc.connect_member(v)
- def c_type(self):
- return c_name(self.name) + pointer_suffix
- def json_type(self):
- return 'value'
- def visit(self, visitor):
- visitor.visit_alternate_type(self.name, self.info, self.variants)
- def is_empty(self):
- return False
- class QAPISchemaCommand(QAPISchemaEntity):
- def __init__(self, name, info, doc, arg_type, ret_type,
- gen, success_response, boxed):
- QAPISchemaEntity.__init__(self, name, info, doc)
- assert not arg_type or isinstance(arg_type, str)
- assert not ret_type or isinstance(ret_type, str)
- self._arg_type_name = arg_type
- self.arg_type = None
- self._ret_type_name = ret_type
- self.ret_type = None
- self.gen = gen
- self.success_response = success_response
- self.boxed = boxed
- def check(self, schema):
- if self._arg_type_name:
- self.arg_type = schema.lookup_type(self._arg_type_name)
- assert (isinstance(self.arg_type, QAPISchemaObjectType) or
- isinstance(self.arg_type, QAPISchemaAlternateType))
- self.arg_type.check(schema)
- if self.boxed:
- if self.arg_type.is_empty():
- raise QAPISemError(self.info,
- "Cannot use 'boxed' with empty type")
- else:
- assert not isinstance(self.arg_type, QAPISchemaAlternateType)
- assert not self.arg_type.variants
- elif self.boxed:
- raise QAPISemError(self.info, "Use of 'boxed' requires 'data'")
- if self._ret_type_name:
- self.ret_type = schema.lookup_type(self._ret_type_name)
- assert isinstance(self.ret_type, QAPISchemaType)
- def visit(self, visitor):
- visitor.visit_command(self.name, self.info,
- self.arg_type, self.ret_type,
- self.gen, self.success_response, self.boxed)
- class QAPISchemaEvent(QAPISchemaEntity):
- def __init__(self, name, info, doc, arg_type, boxed):
- QAPISchemaEntity.__init__(self, name, info, doc)
- assert not arg_type or isinstance(arg_type, str)
- self._arg_type_name = arg_type
- self.arg_type = None
- self.boxed = boxed
- def check(self, schema):
- if self._arg_type_name:
- self.arg_type = schema.lookup_type(self._arg_type_name)
- assert (isinstance(self.arg_type, QAPISchemaObjectType) or
- isinstance(self.arg_type, QAPISchemaAlternateType))
- self.arg_type.check(schema)
- if self.boxed:
- if self.arg_type.is_empty():
- raise QAPISemError(self.info,
- "Cannot use 'boxed' with empty type")
- else:
- assert not isinstance(self.arg_type, QAPISchemaAlternateType)
- assert not self.arg_type.variants
- elif self.boxed:
- raise QAPISemError(self.info, "Use of 'boxed' requires 'data'")
- def visit(self, visitor):
- visitor.visit_event(self.name, self.info, self.arg_type, self.boxed)
- class QAPISchema(object):
- def __init__(self, fname):
- try:
- parser = QAPISchemaParser(open(fname, 'r'))
- self.exprs = check_exprs(parser.exprs)
- self.docs = check_docs(parser.docs)
- self._entity_dict = {}
- self._predefining = True
- self._def_predefineds()
- self._predefining = False
- self._def_exprs()
- self.check()
- except QAPIError as err:
- print >>sys.stderr, err
- exit(1)
- def _def_entity(self, ent):
- # Only the predefined types are allowed to not have info
- assert ent.info or self._predefining
- assert ent.name not in self._entity_dict
- self._entity_dict[ent.name] = ent
- def lookup_entity(self, name, typ=None):
- ent = self._entity_dict.get(name)
- if typ and not isinstance(ent, typ):
- return None
- return ent
- def lookup_type(self, name):
- return self.lookup_entity(name, QAPISchemaType)
- def _def_builtin_type(self, name, json_type, c_type):
- self._def_entity(QAPISchemaBuiltinType(name, json_type, c_type))
- # TODO As long as we have QAPI_TYPES_BUILTIN to share multiple
- # qapi-types.h from a single .c, all arrays of builtins must be
- # declared in the first file whether or not they are used. Nicer
- # would be to use lazy instantiation, while figuring out how to
- # avoid compilation issues with multiple qapi-types.h.
- self._make_array_type(name, None)
- def _def_predefineds(self):
- for t in [('str', 'string', 'char' + pointer_suffix),
- ('number', 'number', 'double'),
- ('int', 'int', 'int64_t'),
- ('int8', 'int', 'int8_t'),
- ('int16', 'int', 'int16_t'),
- ('int32', 'int', 'int32_t'),
- ('int64', 'int', 'int64_t'),
- ('uint8', 'int', 'uint8_t'),
- ('uint16', 'int', 'uint16_t'),
- ('uint32', 'int', 'uint32_t'),
- ('uint64', 'int', 'uint64_t'),
- ('size', 'int', 'uint64_t'),
- ('bool', 'boolean', 'bool'),
- ('any', 'value', 'QObject' + pointer_suffix)]:
- self._def_builtin_type(*t)
- self.the_empty_object_type = QAPISchemaObjectType(
- 'q_empty', None, None, None, [], None)
- self._def_entity(self.the_empty_object_type)
- qtype_values = self._make_enum_members(['none', 'qnull', 'qint',
- 'qstring', 'qdict', 'qlist',
- 'qfloat', 'qbool'])
- self._def_entity(QAPISchemaEnumType('QType', None, None,
- qtype_values, 'QTYPE'))
- def _make_enum_members(self, values):
- return [QAPISchemaMember(v) for v in values]
- def _make_implicit_enum_type(self, name, info, values):
- # See also QAPISchemaObjectTypeMember._pretty_owner()
- name = name + 'Kind' # Use namespace reserved by add_name()
- self._def_entity(QAPISchemaEnumType(
- name, info, None, self._make_enum_members(values), None))
- return name
- def _make_array_type(self, element_type, info):
- name = element_type + 'List' # Use namespace reserved by add_name()
- if not self.lookup_type(name):
- self._def_entity(QAPISchemaArrayType(name, info, element_type))
- return name
- def _make_implicit_object_type(self, name, info, doc, role, members):
- if not members:
- return None
- # See also QAPISchemaObjectTypeMember._pretty_owner()
- name = 'q_obj_%s-%s' % (name, role)
- if not self.lookup_entity(name, QAPISchemaObjectType):
- self._def_entity(QAPISchemaObjectType(name, info, doc, None,
- members, None))
- return name
- def _def_enum_type(self, expr, info, doc):
- name = expr['enum']
- data = expr['data']
- prefix = expr.get('prefix')
- self._def_entity(QAPISchemaEnumType(
- name, info, doc, self._make_enum_members(data), prefix))
- def _make_member(self, name, typ, info):
- optional = False
- if name.startswith('*'):
- name = name[1:]
- optional = True
- if isinstance(typ, list):
- assert len(typ) == 1
- typ = self._make_array_type(typ[0], info)
- return QAPISchemaObjectTypeMember(name, typ, optional)
- def _make_members(self, data, info):
- return [self._make_member(key, value, info)
- for (key, value) in data.iteritems()]
- def _def_struct_type(self, expr, info, doc):
- name = expr['struct']
- base = expr.get('base')
- data = expr['data']
- self._def_entity(QAPISchemaObjectType(name, info, doc, base,
- self._make_members(data, info),
- None))
- def _make_variant(self, case, typ):
- return QAPISchemaObjectTypeVariant(case, typ)
- def _make_simple_variant(self, case, typ, info):
- if isinstance(typ, list):
- assert len(typ) == 1
- typ = self._make_array_type(typ[0], info)
- typ = self._make_implicit_object_type(
- typ, info, None, 'wrapper', [self._make_member('data', typ, info)])
- return QAPISchemaObjectTypeVariant(case, typ)
- def _def_union_type(self, expr, info, doc):
- name = expr['union']
- data = expr['data']
- base = expr.get('base')
- tag_name = expr.get('discriminator')
- tag_member = None
- if isinstance(base, dict):
- base = (self._make_implicit_object_type(
- name, info, doc, 'base', self._make_members(base, info)))
- if tag_name:
- variants = [self._make_variant(key, value)
- for (key, value) in data.iteritems()]
- members = []
- else:
- variants = [self._make_simple_variant(key, value, info)
- for (key, value) in data.iteritems()]
- typ = self._make_implicit_enum_type(name, info,
- [v.name for v in variants])
- tag_member = QAPISchemaObjectTypeMember('type', typ, False)
- members = [tag_member]
- self._def_entity(
- QAPISchemaObjectType(name, info, doc, base, members,
- QAPISchemaObjectTypeVariants(tag_name,
- tag_member,
- variants)))
- def _def_alternate_type(self, expr, info, doc):
- name = expr['alternate']
- data = expr['data']
- variants = [self._make_variant(key, value)
- for (key, value) in data.iteritems()]
- tag_member = QAPISchemaObjectTypeMember('type', 'QType', False)
- self._def_entity(
- QAPISchemaAlternateType(name, info, doc,
- QAPISchemaObjectTypeVariants(None,
- tag_member,
- variants)))
- def _def_command(self, expr, info, doc):
- name = expr['command']
- data = expr.get('data')
- rets = expr.get('returns')
- gen = expr.get('gen', True)
- success_response = expr.get('success-response', True)
- boxed = expr.get('boxed', False)
- if isinstance(data, OrderedDict):
- data = self._make_implicit_object_type(
- name, info, doc, 'arg', self._make_members(data, info))
- if isinstance(rets, list):
- assert len(rets) == 1
- rets = self._make_array_type(rets[0], info)
- self._def_entity(QAPISchemaCommand(name, info, doc, data, rets,
- gen, success_response, boxed))
- def _def_event(self, expr, info, doc):
- name = expr['event']
- data = expr.get('data')
- boxed = expr.get('boxed', False)
- if isinstance(data, OrderedDict):
- data = self._make_implicit_object_type(
- name, info, doc, 'arg', self._make_members(data, info))
- self._def_entity(QAPISchemaEvent(name, info, doc, data, boxed))
- def _def_exprs(self):
- for expr_elem in self.exprs:
- expr = expr_elem['expr']
- info = expr_elem['info']
- doc = expr_elem.get('doc')
- if 'enum' in expr:
- self._def_enum_type(expr, info, doc)
- elif 'struct' in expr:
- self._def_struct_type(expr, info, doc)
- elif 'union' in expr:
- self._def_union_type(expr, info, doc)
- elif 'alternate' in expr:
- self._def_alternate_type(expr, info, doc)
- elif 'command' in expr:
- self._def_command(expr, info, doc)
- elif 'event' in expr:
- self._def_event(expr, info, doc)
- else:
- assert False
- def check(self):
- for ent in self._entity_dict.values():
- ent.check(self)
- def visit(self, visitor):
- visitor.visit_begin(self)
- for (name, entity) in sorted(self._entity_dict.items()):
- if visitor.visit_needed(entity):
- entity.visit(visitor)
- visitor.visit_end()
- #
- # Code generation helpers
- #
- def camel_case(name):
- new_name = ''
- first = True
- for ch in name:
- if ch in ['_', '-']:
- first = True
- elif first:
- new_name += ch.upper()
- first = False
- else:
- new_name += ch.lower()
- return new_name
- # ENUMName -> ENUM_NAME, EnumName1 -> ENUM_NAME1
- # ENUM_NAME -> ENUM_NAME, ENUM_NAME1 -> ENUM_NAME1, ENUM_Name2 -> ENUM_NAME2
- # ENUM24_Name -> ENUM24_NAME
- def camel_to_upper(value):
- c_fun_str = c_name(value, False)
- if value.isupper():
- return c_fun_str
- new_name = ''
- l = len(c_fun_str)
- for i in range(l):
- c = c_fun_str[i]
- # When c is upper and no '_' appears before, do more checks
- if c.isupper() and (i > 0) and c_fun_str[i - 1] != '_':
- if i < l - 1 and c_fun_str[i + 1].islower():
- new_name += '_'
- elif c_fun_str[i - 1].isdigit():
- new_name += '_'
- new_name += c
- return new_name.lstrip('_').upper()
- def c_enum_const(type_name, const_name, prefix=None):
- if prefix is not None:
- type_name = prefix
- return camel_to_upper(type_name) + '_' + c_name(const_name, False).upper()
- c_name_trans = string.maketrans('.-', '__')
- # Map @name to a valid C identifier.
- # If @protect, avoid returning certain ticklish identifiers (like
- # C keywords) by prepending 'q_'.
- #
- # Used for converting 'name' from a 'name':'type' qapi definition
- # into a generated struct member, as well as converting type names
- # into substrings of a generated C function name.
- # '__a.b_c' -> '__a_b_c', 'x-foo' -> 'x_foo'
- # protect=True: 'int' -> 'q_int'; protect=False: 'int' -> 'int'
- def c_name(name, protect=True):
- # ANSI X3J11/88-090, 3.1.1
- c89_words = set(['auto', 'break', 'case', 'char', 'const', 'continue',
- 'default', 'do', 'double', 'else', 'enum', 'extern',
- 'float', 'for', 'goto', 'if', 'int', 'long', 'register',
- 'return', 'short', 'signed', 'sizeof', 'static',
- 'struct', 'switch', 'typedef', 'union', 'unsigned',
- 'void', 'volatile', 'while'])
- # ISO/IEC 9899:1999, 6.4.1
- c99_words = set(['inline', 'restrict', '_Bool', '_Complex', '_Imaginary'])
- # ISO/IEC 9899:2011, 6.4.1
- c11_words = set(['_Alignas', '_Alignof', '_Atomic', '_Generic',
- '_Noreturn', '_Static_assert', '_Thread_local'])
- # GCC http://gcc.gnu.org/onlinedocs/gcc-4.7.1/gcc/C-Extensions.html
- # excluding _.*
- gcc_words = set(['asm', 'typeof'])
- # C++ ISO/IEC 14882:2003 2.11
- cpp_words = set(['bool', 'catch', 'class', 'const_cast', 'delete',
- 'dynamic_cast', 'explicit', 'false', 'friend', 'mutable',
- 'namespace', 'new', 'operator', 'private', 'protected',
- 'public', 'reinterpret_cast', 'static_cast', 'template',
- 'this', 'throw', 'true', 'try', 'typeid', 'typename',
- 'using', 'virtual', 'wchar_t',
- # alternative representations
- 'and', 'and_eq', 'bitand', 'bitor', 'compl', 'not',
- 'not_eq', 'or', 'or_eq', 'xor', 'xor_eq'])
- # namespace pollution:
- polluted_words = set(['unix', 'errno', 'mips', 'sparc'])
- name = name.translate(c_name_trans)
- if protect and (name in c89_words | c99_words | c11_words | gcc_words
- | cpp_words | polluted_words):
- return 'q_' + name
- return name
- eatspace = '\033EATSPACE.'
- pointer_suffix = ' *' + eatspace
- def genindent(count):
- ret = ''
- for _ in range(count):
- ret += ' '
- return ret
- indent_level = 0
- def push_indent(indent_amount=4):
- global indent_level
- indent_level += indent_amount
- def pop_indent(indent_amount=4):
- global indent_level
- indent_level -= indent_amount
- # Generate @code with @kwds interpolated.
- # Obey indent_level, and strip eatspace.
- def cgen(code, **kwds):
- raw = code % kwds
- if indent_level:
- indent = genindent(indent_level)
- # re.subn() lacks flags support before Python 2.7, use re.compile()
- raw = re.subn(re.compile(r'^.', re.MULTILINE),
- indent + r'\g<0>', raw)
- raw = raw[0]
- return re.sub(re.escape(eatspace) + r' *', '', raw)
- def mcgen(code, **kwds):
- if code[0] == '\n':
- code = code[1:]
- return cgen(code, **kwds)
- def guardname(filename):
- return c_name(filename, protect=False).upper()
- def guardstart(name):
- return mcgen('''
- #ifndef %(name)s
- #define %(name)s
- ''',
- name=guardname(name))
- def guardend(name):
- return mcgen('''
- #endif /* %(name)s */
- ''',
- name=guardname(name))
- def gen_enum_lookup(name, values, prefix=None):
- ret = mcgen('''
- const char *const %(c_name)s_lookup[] = {
- ''',
- c_name=c_name(name))
- for value in values:
- index = c_enum_const(name, value, prefix)
- ret += mcgen('''
- [%(index)s] = "%(value)s",
- ''',
- index=index, value=value)
- max_index = c_enum_const(name, '_MAX', prefix)
- ret += mcgen('''
- [%(max_index)s] = NULL,
- };
- ''',
- max_index=max_index)
- return ret
- def gen_enum(name, values, prefix=None):
- # append automatically generated _MAX value
- enum_values = values + ['_MAX']
- ret = mcgen('''
- typedef enum %(c_name)s {
- ''',
- c_name=c_name(name))
- i = 0
- for value in enum_values:
- ret += mcgen('''
- %(c_enum)s = %(i)d,
- ''',
- c_enum=c_enum_const(name, value, prefix),
- i=i)
- i += 1
- ret += mcgen('''
- } %(c_name)s;
- ''',
- c_name=c_name(name))
- ret += mcgen('''
- extern const char *const %(c_name)s_lookup[];
- ''',
- c_name=c_name(name))
- return ret
- def gen_params(arg_type, boxed, extra):
- if not arg_type:
- assert not boxed
- return extra
- ret = ''
- sep = ''
- if boxed:
- ret += '%s arg' % arg_type.c_param_type()
- sep = ', '
- else:
- assert not arg_type.variants
- for memb in arg_type.members:
- ret += sep
- sep = ', '
- if memb.optional:
- ret += 'bool has_%s, ' % c_name(memb.name)
- ret += '%s %s' % (memb.type.c_param_type(),
- c_name(memb.name))
- if extra:
- ret += sep + extra
- return ret
- #
- # Common command line parsing
- #
- def parse_command_line(extra_options='', extra_long_options=[]):
- try:
- opts, args = getopt.gnu_getopt(sys.argv[1:],
- 'chp:o:' + extra_options,
- ['source', 'header', 'prefix=',
- 'output-dir='] + extra_long_options)
- except getopt.GetoptError as err:
- print >>sys.stderr, "%s: %s" % (sys.argv[0], str(err))
- sys.exit(1)
- output_dir = ''
- prefix = ''
- do_c = False
- do_h = False
- extra_opts = []
- for oa in opts:
- o, a = oa
- if o in ('-p', '--prefix'):
- match = re.match(r'([A-Za-z_.-][A-Za-z0-9_.-]*)?', a)
- if match.end() != len(a):
- print >>sys.stderr, \
- "%s: 'funny character '%s' in argument of --prefix" \
- % (sys.argv[0], a[match.end()])
- sys.exit(1)
- prefix = a
- elif o in ('-o', '--output-dir'):
- output_dir = a + '/'
- elif o in ('-c', '--source'):
- do_c = True
- elif o in ('-h', '--header'):
- do_h = True
- else:
- extra_opts.append(oa)
- if not do_c and not do_h:
- do_c = True
- do_h = True
- if len(args) != 1:
- print >>sys.stderr, "%s: need exactly one argument" % sys.argv[0]
- sys.exit(1)
- fname = args[0]
- return (fname, output_dir, do_c, do_h, prefix, extra_opts)
- #
- # Generate output files with boilerplate
- #
- def open_output(output_dir, do_c, do_h, prefix, c_file, h_file,
- c_comment, h_comment):
- guard = guardname(prefix + h_file)
- c_file = output_dir + prefix + c_file
- h_file = output_dir + prefix + h_file
- if output_dir:
- try:
- os.makedirs(output_dir)
- except os.error as e:
- if e.errno != errno.EEXIST:
- raise
- def maybe_open(really, name, opt):
- if really:
- return open(name, opt)
- else:
- import StringIO
- return StringIO.StringIO()
- fdef = maybe_open(do_c, c_file, 'w')
- fdecl = maybe_open(do_h, h_file, 'w')
- fdef.write(mcgen('''
- /* AUTOMATICALLY GENERATED, DO NOT MODIFY */
- %(comment)s
- ''',
- comment=c_comment))
- fdecl.write(mcgen('''
- /* AUTOMATICALLY GENERATED, DO NOT MODIFY */
- %(comment)s
- #ifndef %(guard)s
- #define %(guard)s
- ''',
- comment=h_comment, guard=guard))
- return (fdef, fdecl)
- def close_output(fdef, fdecl):
- fdecl.write('''
- #endif
- ''')
- fdecl.close()
- fdef.close()
|