mirrors/qemu
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
289251b033
qemu/scripts/qapi/common.py
Markus Armbruster c615550df3 qapi: Improve source file read error handling 
qapi-gen.py crashes when it can't open the main schema file, and when
it can't read from any schema file.  Lazy.

Change QAPISchema.__init__() to take a file name instead of a file
object.  Move the open code from _include() to __init__(), so it's
used for the main schema file, too.

Move the read into the try for good measure, and rephrase the error
message.

Reporting open or read failure for the main schema file needs a
QAPISourceInfo representing "no source".  Make QAPISourceInfo cope
with fname=None.

Signed-off-by: Markus Armbruster <armbru@redhat.com>
Reviewed-by: Eric Blake <eblake@redhat.com>
Message-Id: <20190927134639.4284-27-armbru@redhat.com>
2019-09-28 17:17:48 +02:00

2574 lines
86 KiB
Python
Raw Blame History

#
# QAPI helper library
#
# Copyright IBM, Corp. 2011
# Copyright (c) 2013-2018 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.
from __future__ import print_function
from contextlib import contextmanager
import copy
import errno
import os
import re
import string
import sys
from collections import OrderedDict
# 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 = []
#
# Parsing the schema into expressions
#
class QAPISourceInfo(object):
def __init__(self, fname, line, parent):
self.fname = fname
self.line = line
self.parent = parent
self.defn_meta = None
self.defn_name = None
def set_defn(self, meta, name):
self.defn_meta = meta
self.defn_name = name
def next_line(self):
info = copy.copy(self)
info.line += 1
return info
def loc(self):
if self.fname is None:
return sys.argv[0]
ret = self.fname
if self.line is not None:
ret += ':%d' % self.line
return ret
def in_defn(self):
if self.defn_name:
return "%s: In %s '%s':\n" % (self.fname,
self.defn_meta, self.defn_name)
return ''
def include_path(self):
ret = ''
parent = self.parent
while parent:
ret = 'In file included from %s:\n' % parent.loc() + ret
parent = parent.parent
return ret
def __str__(self):
return self.include_path() + self.in_defn() + self.loc()
class QAPIError(Exception):
def __init__(self, info, col, msg):
Exception.__init__(self)
self.info = info
self.col = col
self.msg = msg
def __str__(self):
loc = str(self.info)
if self.col is not None:
assert self.info.line is not None
loc += ':%s' % self.col
return 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.info, col, msg)
class QAPISemError(QAPIError):
def __init__(self, info, msg):
QAPIError.__init__(self, info, None, msg)
class QAPIDoc(object):
"""
A documentation comment block, either definition or free-form
Definition documentation blocks consist of
* a body section: one line naming the definition, followed by an
overview (any number of lines)
* argument sections: a description of each argument (for commands
and events) or member (for structs, unions and alternates)
* features sections: a description of each feature flag
* additional (non-argument) sections, possibly tagged
Free-form documentation blocks consist only of a body section.
"""
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.text = ''
def append(self, line):
self.text += line.rstrip() + '\n'
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()
self.features = OrderedDict()
# a list of Section
self.sections = []
# the current section
self._section = self.body
self._append_line = self._append_body_line
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.
The way that the line is dealt with depends on which part of
the documentation we're parsing right now:
* The body section: ._append_line is ._append_body_line
* An argument section: ._append_line is ._append_args_line
* A features section: ._append_line is ._append_features_line
* An additional section: ._append_line is ._append_various_line
"""
line = line[1:]
if not line:
self._append_freeform(line)
return
if line[0] != ' ':
raise QAPIParseError(self._parser, "missing space after #")
line = line[1:]
self._append_line(line)
def end_comment(self):
self._end_section()
@staticmethod
def _is_section_tag(name):
return name in ('Returns:', 'Since:',
# those are often singular or plural
'Note:', 'Notes:',
'Example:', 'Examples:',
'TODO:')
def _append_body_line(self, line):
"""
Process a line of documentation text in the body section.
If this a symbol line and it is the section's first line, this
is a definition documentation block for that symbol.
If it's a definition documentation block, another symbol line
begins the argument section for the argument named by it, and
a section tag begins an additional section. Start that
section and append the line to it.
Else, append the line to the current section.
"""
name = line.split(' ', 1)[0]
# FIXME not nice: things like '# @foo:' and '# @foo: ' aren't
# recognized, and get silently treated as ordinary text
if not self.symbol and not self.body.text 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")
elif self.symbol:
# This is a definition documentation block
if name.startswith('@') and name.endswith(':'):
self._append_line = self._append_args_line
self._append_args_line(line)
elif line == 'Features:':
self._append_line = self._append_features_line
elif self._is_section_tag(name):
self._append_line = self._append_various_line
self._append_various_line(line)
else:
self._append_freeform(line.strip())
else:
# This is a free-form documentation block
self._append_freeform(line.strip())
def _append_args_line(self, line):
"""
Process a line of documentation text in an argument section.
A symbol line begins the next argument section, a section tag
section or a non-indented line after a blank line begins an
additional section. Start that section and append the line to
it.
Else, append the line to the current section.
"""
name = line.split(' ', 1)[0]
if name.startswith('@') and name.endswith(':'):
line = line[len(name)+1:]
self._start_args_section(name[1:-1])
elif self._is_section_tag(name):
self._append_line = self._append_various_line
self._append_various_line(line)
return
elif (self._section.text.endswith('\n\n')
and line and not line[0].isspace()):
if line == 'Features:':
self._append_line = self._append_features_line
else:
self._start_section()
self._append_line = self._append_various_line
self._append_various_line(line)
return
self._append_freeform(line.strip())
def _append_features_line(self, line):
name = line.split(' ', 1)[0]
if name.startswith('@') and name.endswith(':'):
line = line[len(name)+1:]
self._start_features_section(name[1:-1])
elif self._is_section_tag(name):
self._append_line = self._append_various_line
self._append_various_line(line)
return
elif (self._section.text.endswith('\n\n')
and line and not line[0].isspace()):
self._start_section()
self._append_line = self._append_various_line
self._append_various_line(line)
return
self._append_freeform(line.strip())
def _append_various_line(self, line):
"""
Process a line of documentation text in an additional section.
A symbol line is an error.
A section tag begins an additional section. Start that
section and append the line to it.
Else, append the line to the current section.
"""
name = line.split(' ', 1)[0]
if name.startswith('@') and name.endswith(':'):
raise QAPIParseError(self._parser,
"'%s' can't follow '%s' section"
% (name, self.sections[0].name))
elif self._is_section_tag(name):
line = line[len(name)+1:]
self._start_section(name[:-1])
if (not self._section.name or
not self._section.name.startswith('Example')):
line = line.strip()
self._append_freeform(line)
def _start_symbol_section(self, symbols_dict, 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 symbols_dict:
raise QAPIParseError(self._parser,
"'%s' parameter name duplicated" % name)
assert not self.sections
self._end_section()
self._section = QAPIDoc.ArgSection(name)
symbols_dict[name] = self._section
def _start_args_section(self, name):
self._start_symbol_section(self.args, name)
def _start_features_section(self, name):
self._start_symbol_section(self.features, name)
def _start_section(self, name=None):
if name in ('Returns', 'Since') and self.has_section(name):
raise QAPIParseError(self._parser,
"duplicated '%s' section" % name)
self._end_section()
self._section = QAPIDoc.Section(name)
self.sections.append(self._section)
def _end_section(self):
if self._section:
text = self._section.text = self._section.text.strip()
if self._section.name and (not text or text.isspace()):
raise QAPIParseError(
self._parser,
"empty doc section '%s'" % self._section.name)
self._section = None
def _append_freeform(self, line):
match = re.match(r'(@\S+:)', line)
if match:
raise QAPIParseError(self._parser,
"'%s' not allowed in free-form documentation"
% match.group(1))
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)
def check_expr(self, expr):
if self.has_section('Returns') and 'command' not in expr:
raise QAPISemError(self.info,
"'Returns:' is only valid for commands")
def check(self):
bogus = [name for name, section in self.args.items()
if not section.member]
if bogus:
raise QAPISemError(
self.info,
"the following documented members are not in "
"the declaration: %s" % ", ".join(bogus))
class QAPISchemaParser(object):
def __init__(self, fname, previously_included=[], incl_info=None):
previously_included.append(os.path.abspath(fname))
try:
if sys.version_info[0] >= 3:
fp = open(fname, 'r', encoding='utf-8')
else:
fp = open(fname, 'r')
self.src = fp.read()
except IOError as e:
raise QAPISemError(incl_info or QAPISourceInfo(None, None, None),
"can't read %s file '%s': %s"
% ("include" if incl_info else "schema",
fname,
e.strerror))
if self.src == '' or self.src[-1] != '\n':
self.src += '\n'
self.cursor = 0
self.info = QAPISourceInfo(fname, 1, incl_info)
self.line_pos = 0
self.exprs = []
self.docs = []
self.accept()
cur_doc = None
while self.tok is not None:
info = self.info
if self.tok == '#':
self.reject_expr_doc(cur_doc)
cur_doc = self.get_doc(info)
self.docs.append(cur_doc)
continue
expr = self.get_expr(False)
if 'include' in expr:
self.reject_expr_doc(cur_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")
incl_fname = os.path.join(os.path.dirname(fname),
include)
self.exprs.append({'expr': {'include': incl_fname},
'info': info})
exprs_include = self._include(include, info, incl_fname,
previously_included)
if exprs_include:
self.exprs.extend(exprs_include.exprs)
self.docs.extend(exprs_include.docs)
elif "pragma" in expr:
self.reject_expr_doc(cur_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 an object")
for name, value in pragma.items():
self._pragma(name, value, info)
else:
expr_elem = {'expr': expr,
'info': info}
if cur_doc:
if not cur_doc.symbol:
raise QAPISemError(
cur_doc.info, "definition documentation required")
expr_elem['doc'] = cur_doc
self.exprs.append(expr_elem)
cur_doc = None
self.reject_expr_doc(cur_doc)
@staticmethod
def reject_expr_doc(doc):
if doc and doc.symbol:
raise QAPISemError(
doc.info,
"documentation for '%s' is not followed by the definition"
% doc.symbol)
def _include(self, include, info, incl_fname, previously_included):
incl_abs_fname = os.path.abspath(incl_fname)
# catch inclusion cycle
inf = info
while inf:
if incl_abs_fname == os.path.abspath(inf.fname):
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 None
return QAPISchemaParser(incl_fname, previously_included, info)
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 == "'":
# Note: we accept only printable ASCII
string = ''
esc = False
while True:
ch = self.src[self.cursor]
self.cursor += 1
if ch == '\n':
raise QAPIParseError(self, "missing terminating \"'\"")
if esc:
# Note: we recognize only \\ because we have
# no use for funny characters in strings
if ch != '\\':
raise QAPIParseError(self,
"unknown escape \\%s" % ch)
esc = False
elif ch == '\\':
esc = True
continue
elif ch == "'":
self.val = string
return
if ord(ch) < 32 or ord(ch) >= 127:
raise QAPIParseError(
self, "funny character in string")
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.tok == '\n':
if self.cursor == len(self.src):
self.tok = None
return
self.info = self.info.next_line()
self.line_pos = self.cursor
elif not self.tok.isspace():
# Show up to next structural, whitespace or quote
# character
match = re.match('[^[\\]{}:,\\s\'"]+',
self.src[self.cursor-1:])
raise QAPIParseError(self, "stray '%s'" % match.group(0))
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 '{', '[', string, boolean or 'null'")
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")
doc.end_comment()
self.accept()
return doc
else:
doc.append(self.val)
self.accept(False)
raise QAPIParseError(self, "documentation comment must end with '##'")
#
# Check (context-free) schema expression structure
#
# 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_is_str(name, info, source):
if not isinstance(name, str):
raise QAPISemError(info, "%s requires a string name" % source)
def check_name_str(name, info, source,
allow_optional=False, enum_member=False,
permit_upper=False):
global valid_name
membername = name
if allow_optional and name.startswith('*'):
membername = name[1:]
# 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 has an invalid name" % source)
if not permit_upper and name.lower() != name:
raise QAPISemError(
info, "%s uses uppercase in name" % source)
assert not membername.startswith('*')
def check_defn_name_str(name, info, meta):
check_name_str(name, info, meta, permit_upper=True)
if name.endswith('Kind') or name.endswith('List'):
raise QAPISemError(
info, "%s name should not end in '%s'" % (meta, name[-4:]))
def check_if(expr, info, source):
def check_if_str(ifcond, info):
if not isinstance(ifcond, str):
raise QAPISemError(
info,
"'if' condition of %s must be a string or a list of strings"
% source)
if ifcond.strip() == '':
raise QAPISemError(
info,
"'if' condition '%s' of %s makes no sense"
% (ifcond, source))
ifcond = expr.get('if')
if ifcond is None:
return
if isinstance(ifcond, list):
if ifcond == []:
raise QAPISemError(
info, "'if' condition [] of %s is useless" % source)
for elt in ifcond:
check_if_str(elt, info)
else:
check_if_str(ifcond, info)
def check_type(value, info, source,
allow_array=False, allow_dict=False):
if value is None:
return
# Array type
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)
return
# Type name
if isinstance(value, str):
return
# Anonymous type
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 an object or type name" % source)
permit_upper = allow_dict in name_case_whitelist
# value is a dictionary, check that each member is okay
for (key, arg) in value.items():
key_source = "%s member '%s'" % (source, key)
check_name_str(key, info, key_source,
allow_optional=True, permit_upper=permit_upper)
if c_name(key, False) == 'u' or c_name(key, False).startswith('has_'):
raise QAPISemError(info, "%s uses reserved name" % key_source)
check_keys(arg, info, key_source, ['type'], ['if'])
check_if(arg, info, key_source)
normalize_if(arg)
check_type(arg['type'], info, key_source, allow_array=True)
def check_command(expr, info):
args = expr.get('data')
rets = expr.get('returns')
boxed = expr.get('boxed', False)
if boxed and args is None:
raise QAPISemError(info, "'boxed': true requires 'data'")
check_type(args, info, "'data'", allow_dict=not boxed)
check_type(rets, info, "'returns'", allow_array=True)
def check_event(expr, info):
args = expr.get('data')
boxed = expr.get('boxed', False)
if boxed and args is None:
raise QAPISemError(info, "'boxed': true requires 'data'")
check_type(args, info, "'data'", allow_dict=not boxed)
def check_union(expr, info):
name = expr['union']
base = expr.get('base')
discriminator = expr.get('discriminator')
members = expr['data']
if discriminator is None: # simple union
if base is not None:
raise QAPISemError(info, "'base' requires 'discriminator'")
else: # flat union
check_type(base, info, "'base'", allow_dict=name)
if not base:
raise QAPISemError(info, "'discriminator' requires 'base'")
check_name_is_str(discriminator, info, "'discriminator'")
for (key, value) in members.items():
source = "'data' member '%s'" % key
check_name_str(key, info, source)
check_keys(value, info, source, ['type'], ['if'])
check_if(value, info, source)
normalize_if(value)
check_type(value['type'], info, source, allow_array=not base)
def check_alternate(expr, info):
members = expr['data']
if len(members) == 0:
raise QAPISemError(info, "'data' must not be empty")
for (key, value) in members.items():
source = "'data' member '%s'" % key
check_name_str(key, info, source)
check_keys(value, info, source, ['type'], ['if'])
check_if(value, info, source)
normalize_if(value)
check_type(value['type'], info, source)
def check_enum(expr, info):
name = expr['enum']
members = expr['data']
prefix = expr.get('prefix')
if not isinstance(members, list):
raise QAPISemError(info, "'data' must be an array")
if prefix is not None and not isinstance(prefix, str):
raise QAPISemError(info, "'prefix' must be a string")
permit_upper = name in name_case_whitelist
for member in members:
source = "'data' member"
check_keys(member, info, source, ['name'], ['if'])
check_name_is_str(member['name'], info, source)
source = "%s '%s'" % (source, member['name'])
check_name_str(member['name'], info, source,
enum_member=True, permit_upper=permit_upper)
check_if(member, info, source)
normalize_if(member)
def check_struct(expr, info):
name = expr['struct']
members = expr['data']
features = expr.get('features')
check_type(members, info, "'data'", allow_dict=name)
check_type(expr.get('base'), info, "'base'")
if features:
if not isinstance(features, list):
raise QAPISemError(info, "'features' must be an array")
for f in features:
source = "'features' member"
assert isinstance(f, dict)
check_keys(f, info, source, ['name'], ['if'])
check_name_is_str(f['name'], info, source)
source = "%s '%s'" % (source, f['name'])
check_name_str(f['name'], info, source)
check_if(f, info, source)
normalize_if(f)
def check_keys(value, info, source, required, optional):
def pprint(elems):
return ', '.join("'" + e + "'" for e in sorted(elems))
missing = set(required) - set(value)
if missing:
raise QAPISemError(
info,
"%s misses key%s %s"
% (source, 's' if len(missing) > 1 else '',
pprint(missing)))
allowed = set(required + optional)
unknown = set(value) - allowed
if unknown:
raise QAPISemError(
info,
"%s has unknown key%s %s\nValid keys are %s."
% (source, 's' if len(unknown) > 1 else '',
pprint(unknown), pprint(allowed)))
def check_flags(expr, info):
for key in ['gen', 'success-response']:
if key in expr and expr[key] is not False:
raise QAPISemError(
info, "flag '%s' may only use false value" % key)
for key in ['boxed', 'allow-oob', 'allow-preconfig']:
if key in expr and expr[key] is not True:
raise QAPISemError(
info, "flag '%s' may only use true value" % key)
def normalize_enum(expr):
if isinstance(expr['data'], list):
expr['data'] = [m if isinstance(m, dict) else {'name': m}
for m in expr['data']]
def normalize_members(members):
if isinstance(members, OrderedDict):
for key, arg in members.items():
if isinstance(arg, dict):
continue
members[key] = {'type': arg}
def normalize_features(features):
if isinstance(features, list):
features[:] = [f if isinstance(f, dict) else {'name': f}
for f in features]
def normalize_if(expr):
ifcond = expr.get('if')
if isinstance(ifcond, str):
expr['if'] = [ifcond]
def check_exprs(exprs):
for expr_elem in exprs:
expr = expr_elem['expr']
info = expr_elem['info']
doc = expr_elem.get('doc')
if 'include' in expr:
continue
if 'enum' in expr:
meta = 'enum'
elif 'union' in expr:
meta = 'union'
elif 'alternate' in expr:
meta = 'alternate'
elif 'struct' in expr:
meta = 'struct'
elif 'command' in expr:
meta = 'command'
elif 'event' in expr:
meta = 'event'
else:
raise QAPISemError(info, "expression is missing metatype")
name = expr[meta]
check_name_is_str(name, info, "'%s'" % meta)
info.set_defn(meta, name)
check_defn_name_str(name, info, meta)
if doc:
if doc.symbol != name:
raise QAPISemError(
info, "documentation comment is for '%s'" % doc.symbol)
doc.check_expr(expr)
elif doc_required:
raise QAPISemError(info,
"documentation comment required")
if meta == 'enum':
check_keys(expr, info, meta,
['enum', 'data'], ['if', 'prefix'])
normalize_enum(expr)
check_enum(expr, info)
elif meta == 'union':
check_keys(expr, info, meta,
['union', 'data'],
['base', 'discriminator', 'if'])
normalize_members(expr.get('base'))
normalize_members(expr['data'])
check_union(expr, info)
elif meta == 'alternate':
check_keys(expr, info, meta,
['alternate', 'data'], ['if'])
normalize_members(expr['data'])
check_alternate(expr, info)
elif meta == 'struct':
check_keys(expr, info, meta,
['struct', 'data'], ['base', 'if', 'features'])
normalize_members(expr['data'])
normalize_features(expr.get('features'))
check_struct(expr, info)
elif meta == 'command':
check_keys(expr, info, meta,
['command'],
['data', 'returns', 'boxed', 'if',
'gen', 'success-response', 'allow-oob',
'allow-preconfig'])
normalize_members(expr.get('data'))
check_command(expr, info)
elif meta == 'event':
check_keys(expr, info, meta,
['event'], ['data', 'boxed', 'if'])
normalize_members(expr.get('data'))
check_event(expr, info)
else:
assert False, 'unexpected meta type'
normalize_if(expr)
check_if(expr, info, meta)
check_flags(expr, info)
return exprs
#
# Schema compiler frontend
# TODO catching name collisions in generated code would be nice
#
class QAPISchemaEntity(object):
meta = None
def __init__(self, name, info, doc, ifcond=None):
assert name is None or isinstance(name, str)
self.name = name
self._module = None
# 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
self._ifcond = ifcond or []
self._checked = False
def c_name(self):
return c_name(self.name)
def check(self, schema):
assert not self._checked
if self.info:
self._module = os.path.relpath(self.info.fname,
os.path.dirname(schema.fname))
self._checked = True
@property
def ifcond(self):
assert self._checked
return self._ifcond
@property
def module(self):
assert self._checked
return self._module
def is_implicit(self):
return not self.info
def visit(self, visitor):
assert self._checked
def describe(self):
assert self.meta
return "%s '%s'" % (self.meta, self.name)
class QAPISchemaVisitor(object):
def visit_begin(self, schema):
pass
def visit_end(self):
pass
def visit_module(self, fname):
pass
def visit_needed(self, entity):
# Default to visiting everything
return True
def visit_include(self, fname, info):
pass
def visit_builtin_type(self, name, info, json_type):
pass
def visit_enum_type(self, name, info, ifcond, members, prefix):
pass
def visit_array_type(self, name, info, ifcond, element_type):
pass
def visit_object_type(self, name, info, ifcond, base, members, variants,
features):
pass
def visit_object_type_flat(self, name, info, ifcond, members, variants,
features):
pass
def visit_alternate_type(self, name, info, ifcond, variants):
pass
def visit_command(self, name, info, ifcond, arg_type, ret_type, gen,
success_response, boxed, allow_oob, allow_preconfig):
pass
def visit_event(self, name, info, ifcond, arg_type, boxed):
pass
class QAPISchemaInclude(QAPISchemaEntity):
def __init__(self, fname, info):
QAPISchemaEntity.__init__(self, None, info, None)
self.fname = fname
def visit(self, visitor):
QAPISchemaEntity.visit(self, visitor)
visitor.visit_include(self.fname, self.info)
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 = {
'null': 'QTYPE_QNULL',
'string': 'QTYPE_QSTRING',
'number': 'QTYPE_QNUM',
'int': 'QTYPE_QNUM',
'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
def describe(self):
assert self.meta
return "%s type '%s'" % (self.meta, self.name)
class QAPISchemaBuiltinType(QAPISchemaType):
meta = 'built-in'
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):
QAPISchemaType.visit(self, visitor)
visitor.visit_builtin_type(self.name, self.info, self.json_type())
class QAPISchemaEnumType(QAPISchemaType):
meta = 'enum'
def __init__(self, name, info, doc, ifcond, members, prefix):
QAPISchemaType.__init__(self, name, info, doc, ifcond)
for m in members:
assert isinstance(m, QAPISchemaEnumMember)
m.set_defined_in(name)
assert prefix is None or isinstance(prefix, str)
self.members = members
self.prefix = prefix
def check(self, schema):
QAPISchemaType.check(self, schema)
seen = {}
for m in self.members:
m.check_clash(self.info, seen)
if self.doc:
self.doc.connect_member(m)
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 [m.name for m in self.members]
def json_type(self):
return 'string'
def visit(self, visitor):
QAPISchemaType.visit(self, visitor)
visitor.visit_enum_type(self.name, self.info, self.ifcond,
self.members, self.prefix)
class QAPISchemaArrayType(QAPISchemaType):
meta = 'array'
def __init__(self, name, info, element_type):
QAPISchemaType.__init__(self, name, info, None, None)
assert isinstance(element_type, str)
self._element_type_name = element_type
self.element_type = None
def check(self, schema):
QAPISchemaType.check(self, schema)
self.element_type = schema.resolve_type(
self._element_type_name, self.info,
self.info and self.info.defn_meta)
assert not isinstance(self.element_type, QAPISchemaArrayType)
@property
def ifcond(self):
assert self._checked
return self.element_type.ifcond
@property
def module(self):
assert self._checked
return self.element_type.module
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):
QAPISchemaType.visit(self, visitor)
visitor.visit_array_type(self.name, self.info, self.ifcond,
self.element_type)
def describe(self):
assert self.meta
return "%s type ['%s']" % (self.meta, self._element_type_name)
class QAPISchemaObjectType(QAPISchemaType):
def __init__(self, name, info, doc, ifcond,
base, local_members, variants, features):
# 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, ifcond)
self.meta = 'union' if variants else 'struct'
assert base is None or isinstance(base, str)
for m in local_members:
assert isinstance(m, QAPISchemaObjectTypeMember)
m.set_defined_in(name)
if variants is not None:
assert isinstance(variants, QAPISchemaObjectTypeVariants)
variants.set_defined_in(name)
for f in features:
assert isinstance(f, QAPISchemaFeature)
f.set_defined_in(name)
self._base_name = base
self.base = None
self.local_members = local_members
self.variants = variants
self.members = None
self.features = features
def check(self, schema):
# This calls another type T's .check() exactly when the C
# struct emitted by gen_object() contains that T's C struct
# (pointers don't count).
if self.members is not None:
# A previous .check() completed: nothing to do
return
if self._checked:
# Recursed: C struct contains itself
raise QAPISemError(self.info,
"object %s contains itself" % self.name)
QAPISchemaType.check(self, schema)
assert self._checked and self.members is None
seen = OrderedDict()
if self._base_name:
self.base = schema.resolve_type(self._base_name, self.info,
"'base'")
if (not isinstance(self.base, QAPISchemaObjectType)
or self.base.variants):
raise QAPISemError(
self.info,
"'base' requires a struct type, %s isn't"
% self.base.describe())
self.base.check(schema)
self.base.check_clash(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)
members = seen.values()
if self.variants:
self.variants.check(schema, seen)
self.variants.check_clash(self.info, seen)
# Features are in a name space separate from members
seen = {}
for f in self.features:
f.check_clash(self.info, seen)
if self.doc:
self.doc.check()
self.members = members # mark completed
# 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, info, seen):
assert self._checked
assert not self.variants # not implemented
for m in self.members:
m.check_clash(info, seen)
@property
def ifcond(self):
assert self._checked
if isinstance(self._ifcond, QAPISchemaType):
# Simple union wrapper type inherits from wrapped type;
# see _make_implicit_object_type()
return self._ifcond.ifcond
return self._ifcond
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):
QAPISchemaType.visit(self, visitor)
visitor.visit_object_type(self.name, self.info, self.ifcond,
self.base, self.local_members, self.variants,
self.features)
visitor.visit_object_type_flat(self.name, self.info, self.ifcond,
self.members, self.variants,
self.features)
class QAPISchemaMember(object):
""" Represents object members, enum members and features """
role = 'member'
def __init__(self, name, info, ifcond=None):
assert isinstance(name, str)
self.name = name
self.info = info
self.ifcond = ifcond or []
self.defined_in = None
def set_defined_in(self, name):
assert not self.defined_in
self.defined_in = name
def check_clash(self, info, seen):
cname = c_name(self.name)
if cname in seen:
raise QAPISemError(
info,
"%s collides with %s"
% (self.describe(info), seen[cname].describe(info)))
seen[cname] = self
def describe(self, info):
role = self.role
defined_in = self.defined_in
assert defined_in
if defined_in.startswith('q_obj_'):
# See QAPISchema._make_implicit_object_type() - reverse the
# mapping there to create a nice human-readable description
defined_in = defined_in[6:]
if defined_in.endswith('-arg'):
# Implicit type created for a command's dict 'data'
assert role == 'member'
role = 'parameter'
elif defined_in.endswith('-base'):
# Implicit type created for a flat union's dict 'base'
role = 'base ' + role
else:
# Implicit type created for a simple union's branch
assert defined_in.endswith('-wrapper')
# Unreachable and not implemented
assert False
elif defined_in.endswith('Kind'):
# See QAPISchema._make_implicit_enum_type()
# Implicit enum created for simple union's branches
assert role == 'value'
role = 'branch'
elif defined_in != info.defn_name:
return "%s '%s' of type '%s'" % (role, self.name, defined_in)
return "%s '%s'" % (role, self.name)
class QAPISchemaEnumMember(QAPISchemaMember):
role = 'value'
class QAPISchemaFeature(QAPISchemaMember):
role = 'feature'
class QAPISchemaObjectTypeMember(QAPISchemaMember):
def __init__(self, name, info, typ, optional, ifcond=None):
QAPISchemaMember.__init__(self, name, info, ifcond)
assert isinstance(typ, str)
assert isinstance(optional, bool)
self._type_name = typ
self.type = None
self.optional = optional
def check(self, schema):
assert self.defined_in
self.type = schema.resolve_type(self._type_name, self.info,
self.describe)
class QAPISchemaObjectTypeVariants(object):
def __init__(self, tag_name, info, 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))
for v in variants:
assert isinstance(v, QAPISchemaObjectTypeVariant)
self._tag_name = tag_name
self.info = info
self.tag_member = tag_member
self.variants = variants
def set_defined_in(self, name):
for v in self.variants:
v.set_defined_in(name)
def check(self, schema, seen):
if not self.tag_member: # flat union
self.tag_member = seen.get(c_name(self._tag_name))
base = "'base'"
# Pointing to the base type when not implicit would be
# nice, but we don't know it here
if not self.tag_member or self._tag_name != self.tag_member.name:
raise QAPISemError(
self.info,
"discriminator '%s' is not a member of %s"
% (self._tag_name, base))
# Here we do:
base_type = schema.lookup_type(self.tag_member.defined_in)
assert base_type
if not base_type.is_implicit():
base = "base type '%s'" % self.tag_member.defined_in
if not isinstance(self.tag_member.type, QAPISchemaEnumType):
raise QAPISemError(
self.info,
"discriminator member '%s' of %s must be of enum type"
% (self._tag_name, base))
if self.tag_member.optional:
raise QAPISemError(
self.info,
"discriminator member '%s' of %s must not be optional"
% (self._tag_name, base))
if self.tag_member.ifcond:
raise QAPISemError(
self.info,
"discriminator member '%s' of %s must not be conditional"
% (self._tag_name, base))
else: # simple union
assert isinstance(self.tag_member.type, QAPISchemaEnumType)
assert not self.tag_member.optional
assert self.tag_member.ifcond == []
if self._tag_name: # flat union
# branches that are not explicitly covered get an empty type
cases = set([v.name for v in self.variants])
for m in self.tag_member.type.members:
if m.name not in cases:
v = QAPISchemaObjectTypeVariant(m.name, self.info,
'q_empty', m.ifcond)
v.set_defined_in(self.tag_member.defined_in)
self.variants.append(v)
if not self.variants:
raise QAPISemError(self.info, "union has no branches")
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:
if v.name not in self.tag_member.type.member_names():
raise QAPISemError(
self.info,
"branch '%s' is not a value of %s"
% (v.name, self.tag_member.type.describe()))
if (not isinstance(v.type, QAPISchemaObjectType)
or v.type.variants):
raise QAPISemError(
self.info,
"%s cannot use %s"
% (v.describe(self.info), v.type.describe()))
v.type.check(schema)
def check_clash(self, info, seen):
for v in self.variants:
# Reset seen map for each variant, since qapi names from one
# branch do not affect another branch
v.type.check_clash(info, dict(seen))
class QAPISchemaObjectTypeVariant(QAPISchemaObjectTypeMember):
role = 'branch'
def __init__(self, name, info, typ, ifcond=None):
QAPISchemaObjectTypeMember.__init__(self, name, info, typ,
False, ifcond)
class QAPISchemaAlternateType(QAPISchemaType):
meta = 'alternate'
def __init__(self, name, info, doc, ifcond, variants):
QAPISchemaType.__init__(self, name, info, doc, ifcond)
assert isinstance(variants, QAPISchemaObjectTypeVariants)
assert variants.tag_member
variants.set_defined_in(name)
variants.tag_member.set_defined_in(self.name)
self.variants = variants
def check(self, schema):
QAPISchemaType.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 = {}
types_seen = {}
for v in self.variants.variants:
v.check_clash(self.info, seen)
qtype = v.type.alternate_qtype()
if not qtype:
raise QAPISemError(
self.info,
"%s cannot use %s"
% (v.describe(self.info), v.type.describe()))
conflicting = set([qtype])
if qtype == 'QTYPE_QSTRING':
if isinstance(v.type, QAPISchemaEnumType):
for m in v.type.members:
if m.name in ['on', 'off']:
conflicting.add('QTYPE_QBOOL')
if re.match(r'[-+0-9.]', m.name):
# lazy, could be tightened
conflicting.add('QTYPE_QNUM')
else:
conflicting.add('QTYPE_QNUM')
conflicting.add('QTYPE_QBOOL')
for qt in conflicting:
if qt in types_seen:
raise QAPISemError(
self.info,
"%s can't be distinguished from '%s'"
% (v.describe(self.info), types_seen[qt]))
types_seen[qt] = v.name
if self.doc:
self.doc.connect_member(v)
if self.doc:
self.doc.check()
def c_type(self):
return c_name(self.name) + pointer_suffix
def json_type(self):
return 'value'
def visit(self, visitor):
QAPISchemaType.visit(self, visitor)
visitor.visit_alternate_type(self.name, self.info, self.ifcond,
self.variants)
class QAPISchemaCommand(QAPISchemaEntity):
meta = 'command'
def __init__(self, name, info, doc, ifcond, arg_type, ret_type,
gen, success_response, boxed, allow_oob, allow_preconfig):
QAPISchemaEntity.__init__(self, name, info, doc, ifcond)
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
self.allow_oob = allow_oob
self.allow_preconfig = allow_preconfig
def check(self, schema):
QAPISchemaEntity.check(self, schema)
if self._arg_type_name:
self.arg_type = schema.resolve_type(
self._arg_type_name, self.info, "command's 'data'")
if not isinstance(self.arg_type, QAPISchemaObjectType):
raise QAPISemError(
self.info,
"command's 'data' cannot take %s"
% self.arg_type.describe())
if self.arg_type.variants and not self.boxed:
raise QAPISemError(
self.info,
"command's 'data' can take %s only with 'boxed': true"
% self.arg_type.describe())
if self._ret_type_name:
self.ret_type = schema.resolve_type(
self._ret_type_name, self.info, "command's 'returns'")
if self.name not in returns_whitelist:
if not (isinstance(self.ret_type, QAPISchemaObjectType)
or (isinstance(self.ret_type, QAPISchemaArrayType)
and isinstance(self.ret_type.element_type,
QAPISchemaObjectType))):
raise QAPISemError(
self.info,
"command's 'returns' cannot take %s"
% self.ret_type.describe())
def visit(self, visitor):
QAPISchemaEntity.visit(self, visitor)
visitor.visit_command(self.name, self.info, self.ifcond,
self.arg_type, self.ret_type,
self.gen, self.success_response,
self.boxed, self.allow_oob,
self.allow_preconfig)
class QAPISchemaEvent(QAPISchemaEntity):
meta = 'event'
def __init__(self, name, info, doc, ifcond, arg_type, boxed):
QAPISchemaEntity.__init__(self, name, info, doc, ifcond)
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):
QAPISchemaEntity.check(self, schema)
if self._arg_type_name:
self.arg_type = schema.resolve_type(
self._arg_type_name, self.info, "event's 'data'")
if not isinstance(self.arg_type, QAPISchemaObjectType):
raise QAPISemError(
self.info,
"event's 'data' cannot take %s"
% self.arg_type.describe())
if self.arg_type.variants and not self.boxed:
raise QAPISemError(
self.info,
"event's 'data' can take %s only with 'boxed': true"
% self.arg_type.describe())
def visit(self, visitor):
QAPISchemaEntity.visit(self, visitor)
visitor.visit_event(self.name, self.info, self.ifcond,
self.arg_type, self.boxed)
class QAPISchema(object):
def __init__(self, fname):
self.fname = fname
parser = QAPISchemaParser(fname)
exprs = check_exprs(parser.exprs)
self.docs = parser.docs
self._entity_list = []
self._entity_dict = {}
self._predefining = True
self._def_predefineds()
self._predefining = False
self._def_exprs(exprs)
self.check()
def _def_entity(self, ent):
# Only the predefined types are allowed to not have info
assert ent.info or self._predefining
self._entity_list.append(ent)
if ent.name is None:
return
# TODO reject names that differ only in '_' vs. '.' vs. '-',
# because they're liable to clash in generated C.
other_ent = self._entity_dict.get(ent.name)
if other_ent:
if other_ent.info:
where = QAPIError(other_ent.info, None, "previous definition")
raise QAPISemError(
ent.info,
"'%s' is already defined\n%s" % (ent.name, where))
raise QAPISemError(
ent.info, "%s is already defined" % other_ent.describe())
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 resolve_type(self, name, info, what):
typ = self.lookup_type(name)
if not typ:
if callable(what):
what = what(info)
raise QAPISemError(
info, "%s uses unknown type '%s'" % (what, name))
return typ
def _def_include(self, expr, info, doc):
include = expr['include']
assert doc is None
main_info = info
while main_info.parent:
main_info = main_info.parent
fname = os.path.relpath(include, os.path.dirname(main_info.fname))
self._def_entity(QAPISchemaInclude(fname, info))
def _def_builtin_type(self, name, json_type, c_type):
self._def_entity(QAPISchemaBuiltinType(name, json_type, c_type))
# Instantiating only the arrays that are actually used would
# be nice, but we can't as long as their generated code
# (qapi-builtin-types.[ch]) may be shared by some other
# schema.
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),
('null', 'null', 'QNull' + pointer_suffix)]:
self._def_builtin_type(*t)
self.the_empty_object_type = QAPISchemaObjectType(
'q_empty', None, None, None, None, [], None, [])
self._def_entity(self.the_empty_object_type)
qtypes = ['none', 'qnull', 'qnum', 'qstring', 'qdict', 'qlist',
'qbool']
qtype_values = self._make_enum_members(
[{'name': n} for n in qtypes], None)
self._def_entity(QAPISchemaEnumType('QType', None, None, None,
qtype_values, 'QTYPE'))
def _make_features(self, features, info):
return [QAPISchemaFeature(f['name'], info, f.get('if'))
for f in features]
def _make_enum_members(self, values, info):
return [QAPISchemaEnumMember(v['name'], info, v.get('if'))
for v in values]
def _make_implicit_enum_type(self, name, info, ifcond, values):
# See also QAPISchemaObjectTypeMember.describe()
name = name + 'Kind' # reserved by check_defn_name_str()
self._def_entity(QAPISchemaEnumType(
name, info, None, ifcond, self._make_enum_members(values, info),
None))
return name
def _make_array_type(self, element_type, info):
name = element_type + 'List' # reserved by check_defn_name_str()
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, ifcond,
role, members):
if not members:
return None
# See also QAPISchemaObjectTypeMember.describe()
name = 'q_obj_%s-%s' % (name, role)
typ = self.lookup_entity(name, QAPISchemaObjectType)
if typ:
# The implicit object type has multiple users. This can
# happen only for simple unions' implicit wrapper types.
# Its ifcond should be the disjunction of its user's
# ifconds. Not implemented. Instead, we always pass the
# wrapped type's ifcond, which is trivially the same for all
# users. It's also necessary for the wrapper to compile.
# But it's not tight: the disjunction need not imply it. We
# may end up compiling useless wrapper types.
# TODO kill simple unions or implement the disjunction
assert (ifcond or []) == typ._ifcond # pylint: disable=protected-access
else:
self._def_entity(QAPISchemaObjectType(name, info, doc, ifcond,
None, members, None, []))
return name
def _def_enum_type(self, expr, info, doc):
name = expr['enum']
data = expr['data']
prefix = expr.get('prefix')
ifcond = expr.get('if')
self._def_entity(QAPISchemaEnumType(
name, info, doc, ifcond,
self._make_enum_members(data, info), prefix))
def _make_member(self, name, typ, ifcond, 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, info, typ, optional, ifcond)
def _make_members(self, data, info):
return [self._make_member(key, value['type'], value.get('if'), info)
for (key, value) in data.items()]
def _def_struct_type(self, expr, info, doc):
name = expr['struct']
base = expr.get('base')
data = expr['data']
ifcond = expr.get('if')
features = expr.get('features', [])
self._def_entity(QAPISchemaObjectType(
name, info, doc, ifcond, base,
self._make_members(data, info),
None,
self._make_features(features, info)))
def _make_variant(self, case, typ, ifcond, info):
return QAPISchemaObjectTypeVariant(case, info, typ, ifcond)
def _make_simple_variant(self, case, typ, ifcond, 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, self.lookup_type(typ),
'wrapper', [self._make_member('data', typ, None, info)])
return QAPISchemaObjectTypeVariant(case, info, typ, ifcond)
def _def_union_type(self, expr, info, doc):
name = expr['union']
data = expr['data']
base = expr.get('base')
ifcond = expr.get('if')
tag_name = expr.get('discriminator')
tag_member = None
if isinstance(base, dict):
base = self._make_implicit_object_type(
name, info, doc, ifcond,
'base', self._make_members(base, info))
if tag_name:
variants = [self._make_variant(key, value['type'],
value.get('if'), info)
for (key, value) in data.items()]
members = []
else:
variants = [self._make_simple_variant(key, value['type'],
value.get('if'), info)
for (key, value) in data.items()]
enum = [{'name': v.name, 'if': v.ifcond} for v in variants]
typ = self._make_implicit_enum_type(name, info, ifcond, enum)
tag_member = QAPISchemaObjectTypeMember('type', info, typ, False)
members = [tag_member]
self._def_entity(
QAPISchemaObjectType(name, info, doc, ifcond, base, members,
QAPISchemaObjectTypeVariants(
tag_name, info, tag_member, variants),
[]))
def _def_alternate_type(self, expr, info, doc):
name = expr['alternate']
data = expr['data']
ifcond = expr.get('if')
variants = [self._make_variant(key, value['type'], value.get('if'),
info)
for (key, value) in data.items()]
tag_member = QAPISchemaObjectTypeMember('type', info, 'QType', False)
self._def_entity(
QAPISchemaAlternateType(name, info, doc, ifcond,
QAPISchemaObjectTypeVariants(
None, info, 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)
allow_oob = expr.get('allow-oob', False)
allow_preconfig = expr.get('allow-preconfig', False)
ifcond = expr.get('if')
if isinstance(data, OrderedDict):
data = self._make_implicit_object_type(
name, info, doc, ifcond, '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, ifcond, data, rets,
gen, success_response,
boxed, allow_oob, allow_preconfig))
def _def_event(self, expr, info, doc):
name = expr['event']
data = expr.get('data')
boxed = expr.get('boxed', False)
ifcond = expr.get('if')
if isinstance(data, OrderedDict):
data = self._make_implicit_object_type(
name, info, doc, ifcond, 'arg', self._make_members(data, info))
self._def_entity(QAPISchemaEvent(name, info, doc, ifcond, data, boxed))
def _def_exprs(self, exprs):
for expr_elem in 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)
elif 'include' in expr:
self._def_include(expr, info, doc)
else:
assert False
def check(self):
for ent in self._entity_list:
ent.check(self)
def visit(self, visitor):
visitor.visit_begin(self)
module = None
visitor.visit_module(module)
for entity in self._entity_list:
if visitor.visit_needed(entity):
if entity.module != module:
module = entity.module
visitor.visit_module(module)
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 = ''
length = len(c_fun_str)
for i in range(length):
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 < length - 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()
if hasattr(str, 'maketrans'):
c_name_trans = str.maketrans('.-', '__')
else:
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', 'i386'])
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, 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 c_fname(filename):
return re.sub(r'[^A-Za-z0-9_]', '_', filename)
def guardstart(name):
return mcgen('''
#ifndef %(name)s
#define %(name)s
''',
name=c_fname(name).upper())
def guardend(name):
return mcgen('''
#endif /* %(name)s */
''',
name=c_fname(name).upper())
def gen_if(ifcond):
ret = ''
for ifc in ifcond:
ret += mcgen('''
#if %(cond)s
''', cond=ifc)
return ret
def gen_endif(ifcond):
ret = ''
for ifc in reversed(ifcond):
ret += mcgen('''
#endif /* %(cond)s */
''', cond=ifc)
return ret
def _wrap_ifcond(ifcond, before, after):
if before == after:
return after # suppress empty #if ... #endif
assert after.startswith(before)
out = before
added = after[len(before):]
if added[0] == '\n':
out += '\n'
added = added[1:]
out += gen_if(ifcond)
out += added
out += gen_endif(ifcond)
return out
def gen_enum_lookup(name, members, prefix=None):
ret = mcgen('''
const QEnumLookup %(c_name)s_lookup = {
.array = (const char *const[]) {
''',
c_name=c_name(name))
for m in members:
ret += gen_if(m.ifcond)
index = c_enum_const(name, m.name, prefix)
ret += mcgen('''
[%(index)s] = "%(name)s",
''',
index=index, name=m.name)
ret += gen_endif(m.ifcond)
ret += mcgen('''
},
.size = %(max_index)s
};
''',
max_index=c_enum_const(name, '_MAX', prefix))
return ret
def gen_enum(name, members, prefix=None):
# append automatically generated _MAX value
enum_members = members + [QAPISchemaEnumMember('_MAX', None)]
ret = mcgen('''
typedef enum %(c_name)s {
''',
c_name=c_name(name))
for m in enum_members:
ret += gen_if(m.ifcond)
ret += mcgen('''
%(c_enum)s,
''',
c_enum=c_enum_const(name, m.name, prefix))
ret += gen_endif(m.ifcond)
ret += mcgen('''
} %(c_name)s;
''',
c_name=c_name(name))
ret += mcgen('''
#define %(c_name)s_str(val) \\
qapi_enum_lookup(&%(c_name)s_lookup, (val))
extern const QEnumLookup %(c_name)s_lookup;
''',
c_name=c_name(name))
return ret
def build_params(arg_type, boxed, extra=None):
ret = ''
sep = ''
if boxed:
assert arg_type
ret += '%s arg' % arg_type.c_param_type()
sep = ', '
elif arg_type:
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 if ret else 'void'
#
# Accumulate and write output
#
class QAPIGen(object):
def __init__(self, fname):
self.fname = fname
self._preamble = ''
self._body = ''
def preamble_add(self, text):
self._preamble += text
def add(self, text):
self._body += text
def get_content(self):
return self._top() + self._preamble + self._body + self._bottom()
def _top(self):
return ''
def _bottom(self):
return ''
def write(self, output_dir):
pathname = os.path.join(output_dir, self.fname)
dir = os.path.dirname(pathname)
if dir:
try:
os.makedirs(dir)
except os.error as e:
if e.errno != errno.EEXIST:
raise
fd = os.open(pathname, os.O_RDWR | os.O_CREAT, 0o666)
if sys.version_info[0] >= 3:
f = open(fd, 'r+', encoding='utf-8')
else:
f = os.fdopen(fd, 'r+')
text = self.get_content()
oldtext = f.read(len(text) + 1)
if text != oldtext:
f.seek(0)
f.truncate(0)
f.write(text)
f.close()
@contextmanager
def ifcontext(ifcond, *args):
"""A 'with' statement context manager to wrap with start_if()/end_if()
*args: any number of QAPIGenCCode
Example::
with ifcontext(ifcond, self._genh, self._genc):
modify self._genh and self._genc ...
Is equivalent to calling::
self._genh.start_if(ifcond)
self._genc.start_if(ifcond)
modify self._genh and self._genc ...
self._genh.end_if()
self._genc.end_if()
"""
for arg in args:
arg.start_if(ifcond)
yield
for arg in args:
arg.end_if()
class QAPIGenCCode(QAPIGen):
def __init__(self, fname):
QAPIGen.__init__(self, fname)
self._start_if = None
def start_if(self, ifcond):
assert self._start_if is None
self._start_if = (ifcond, self._body, self._preamble)
def end_if(self):
assert self._start_if
self._wrap_ifcond()
self._start_if = None
def _wrap_ifcond(self):
self._body = _wrap_ifcond(self._start_if[0],
self._start_if[1], self._body)
self._preamble = _wrap_ifcond(self._start_if[0],
self._start_if[2], self._preamble)
def get_content(self):
assert self._start_if is None
return QAPIGen.get_content(self)
class QAPIGenC(QAPIGenCCode):
def __init__(self, fname, blurb, pydoc):
QAPIGenCCode.__init__(self, fname)
self._blurb = blurb
self._copyright = '\n * '.join(re.findall(r'^Copyright .*', pydoc,
re.MULTILINE))
def _top(self):
return mcgen('''
/* AUTOMATICALLY GENERATED, DO NOT MODIFY */
/*
%(blurb)s
*
* %(copyright)s
*
* This work is licensed under the terms of the GNU LGPL, version 2.1 or later.
* See the COPYING.LIB file in the top-level directory.
*/
''',
blurb=self._blurb, copyright=self._copyright)
def _bottom(self):
return mcgen('''
/* Dummy declaration to prevent empty .o file */
char qapi_dummy_%(name)s;
''',
name=c_fname(self.fname))
class QAPIGenH(QAPIGenC):
def _top(self):
return QAPIGenC._top(self) + guardstart(self.fname)
def _bottom(self):
return guardend(self.fname)
class QAPIGenDoc(QAPIGen):
def _top(self):
return (QAPIGen._top(self)
+ '@c AUTOMATICALLY GENERATED, DO NOT MODIFY\n\n')
class QAPISchemaMonolithicCVisitor(QAPISchemaVisitor):
def __init__(self, prefix, what, blurb, pydoc):
self._prefix = prefix
self._what = what
self._genc = QAPIGenC(self._prefix + self._what + '.c',
blurb, pydoc)
self._genh = QAPIGenH(self._prefix + self._what + '.h',
blurb, pydoc)
def write(self, output_dir):
self._genc.write(output_dir)
self._genh.write(output_dir)
class QAPISchemaModularCVisitor(QAPISchemaVisitor):
def __init__(self, prefix, what, blurb, pydoc):
self._prefix = prefix
self._what = what
self._blurb = blurb
self._pydoc = pydoc
self._genc = None
self._genh = None
self._module = {}
self._main_module = None
@staticmethod
def _is_user_module(name):
return name and not name.startswith('./')
@staticmethod
def _is_builtin_module(name):
return not name
def _module_dirname(self, what, name):
if self._is_user_module(name):
return os.path.dirname(name)
return ''
def _module_basename(self, what, name):
ret = '' if self._is_builtin_module(name) else self._prefix
if self._is_user_module(name):
basename = os.path.basename(name)
ret += what
if name != self._main_module:
ret += '-' + os.path.splitext(basename)[0]
else:
name = name[2:] if name else 'builtin'
ret += re.sub(r'-', '-' + name + '-', what)
return ret
def _module_filename(self, what, name):
return os.path.join(self._module_dirname(what, name),
self._module_basename(what, name))
def _add_module(self, name, blurb):
basename = self._module_filename(self._what, name)
genc = QAPIGenC(basename + '.c', blurb, self._pydoc)
genh = QAPIGenH(basename + '.h', blurb, self._pydoc)
self._module[name] = (genc, genh)
self._set_module(name)
def _add_user_module(self, name, blurb):
assert self._is_user_module(name)
if self._main_module is None:
self._main_module = name
self._add_module(name, blurb)
def _add_system_module(self, name, blurb):
self._add_module(name and './' + name, blurb)
def _set_module(self, name):
self._genc, self._genh = self._module[name]
def write(self, output_dir, opt_builtins=False):
for name in self._module:
if self._is_builtin_module(name) and not opt_builtins:
continue
(genc, genh) = self._module[name]
genc.write(output_dir)
genh.write(output_dir)
def _begin_user_module(self, name):
pass
def visit_module(self, name):
if name in self._module:
self._set_module(name)
elif self._is_builtin_module(name):
# The built-in module has not been created. No code may
# be generated.
self._genc = None
self._genh = None
else:
self._add_user_module(name, self._blurb)
self._begin_user_module(name)
def visit_include(self, name, info):
relname = os.path.relpath(self._module_filename(self._what, name),
os.path.dirname(self._genh.fname))
self._genh.preamble_add(mcgen('''
#include "%(relname)s.h"
''',
relname=relname))
Reference in New Issue View Git Blame Copy Permalink