mkatanbaf commented on code in PR #12125:
URL: https://github.com/apache/tvm/pull/12125#discussion_r943948149
##########
apps/microtvm/zephyr/template_project/microtvm_api_server.py:
##########
@@ -904,5 +982,192 @@ def _wait_for_qemu(self):
raise ValueError(f"{item} not expected.")
+class ZephyrFvpMakeResult(enum.Enum):
+ FVP_STARTED = "fvp_started"
+ MICROTVM_API_SERVER_INIT = "fvp_initialized"
+ MAKE_FAILED = "make_failed"
+ EOF = "eof"
+
+
+class BlockingStream:
+ """Reimplementation of Stream class from Iris with blocking semantics."""
+
+ def __init__(self):
+ self.q = queue.Queue()
+ self.unread = None
+
+ def read(self, n=-1, timeout_sec=None):
+ assert (
+ n != -1
+ ), "expect firmware to open stdin using raw mode, and therefore expect
sized read requests"
+
+ data = b""
+ if self.unread:
+ data = data + self.unread
+ self.unread = None
+
+ while len(data) < n:
+ try:
+ # When there is some data to return, fetch as much as
possible, then return what we can.
+ # When there is no data yet to return, block.
+ data += self.q.get(block=not len(data), timeout=timeout_sec)
+ except queue.Empty:
+ break
+
+ if len(data) > n:
+ self.unread = data[n:]
+ data = data[:n]
+
+ return data
+
+ readline = read
+
+ def write(self, data):
+ self.q.put(data)
+
+
+class ZephyrFvpTransport:
+ """A transport class that communicates with the ARM FVP via Iris server."""
+
+ def __init__(self, options):
+ self.options = options
+ self.proc = None
+ self._queue = queue.Queue()
+ self._import_iris()
+
+ def _import_iris(self):
+ # Location as seen in the FVP_Corstone_SSE-300_11.15_24 tar.
+ iris_lib_path = (
+ pathlib.Path(self.options["arm_fvp_path"]).parent.parent.parent
+ / "Iris"
+ / "Python"
+ / "iris"
+ )
+
+ sys.path.insert(0, str(iris_lib_path.parent))
+ try:
+ import iris.NetworkModelInitializer
+ finally:
+ sys.path.pop(0)
+
+ self._iris_lib = iris
+
+ def _convertStringToU64Array(strValue):
+ numBytes = len(strValue)
+ if numBytes == 0:
+ return []
+
+ numU64 = (numBytes + 7) // 8
+ # Extend the string ending with '\0', so that the string length is
multiple of 8.
+ # E.g. 'hello' is extended to: 'hello'+\0\0\0
+ strExt = strValue.ljust(8 * numU64, b"\0")
+ # Convert the string to a list of uint64_t in little endian
+ return struct.unpack("<{}Q".format(numU64), strExt)
+
+ iris.iris.convertStringToU64Array = _convertStringToU64Array
+
+ def open(self):
+ args = ["ninja"]
+ if self.options.get("verbose"):
+ args.append("-v")
+ args.append("run")
+ env = dict(os.environ)
+ env["FVP_BIN_PATH"] =
str(pathlib.Path(self.options["arm_fvp_path"]).parent)
Review Comment:
regarding the first point,you are correct to say that the option is required
in open, but only for the FVP projects. (we still need the `arm_fvp_path` in
open transport in `_import_iris` function. ) As you suggested, I added an
assert there to make it clear.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
