FPGA/myhdl
1
0
Fork 0
mirror of https://github.com/myhdl/myhdl.git synced 2025-01-24 21:52:56 +08:00
Code Issues Projects Releases Wiki Activity

use subprocess instead of forking for cosim

Browse Source
This commit is contained in:
Keerthan Jaic 2015-04-08 16:49:58 -04:00
parent d65bdbfd72
commit 5fa590a632
3 changed files with 63 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

119
myhdl/_Cosimulation.py
View File

@ -21,12 +21,13 @@
from __future__ import absolute_import from __future__ import absolute_import
import sys
import os import os
import shlex
import subprocess
from myhdl._intbv import intbv from myhdl._intbv import intbv
from myhdl import _simulator, CosimulationError from myhdl import _simulator, CosimulationError
from myhdl._compat import PY2, to_bytes, to_str from myhdl._compat import PY2, string_types, to_bytes, to_str
_MAXLINE = 4096 _MAXLINE = 4096
@ -67,67 +68,69 @@ class Cosimulation(object):
self._toSizes = toSizes = [] self._toSizes = toSizes = []
self._toSigs = toSigs = [] self._toSigs = toSigs = []
self._toSigDict = toSigDict = {} self._toSigDict = toSigDict = {}
self._hasChange = 0 self._hasChange = 0
self._getMode = 1 self._getMode = 1
child_pid = self._child_pid = os.fork() def close_rt_wf():
if child_pid == 0:
os.close(rt) os.close(rt)
os.close(wf) os.close(wf)
os.environ['MYHDL_TO_PIPE'] = str(wt)
os.environ['MYHDL_FROM_PIPE'] = str(rf) env = os.environ.copy()
if isinstance(exe, list): arglist = exe env['MYHDL_TO_PIPE'] = str(wt)
else: arglist = exe.split() env['MYHDL_FROM_PIPE'] = str(rf)
p = arglist[0]
arglist[0] = os.path.basename(p) if isinstance(exe, string_types):
try: exe = shlex.split(exe)
os.execvp(p, arglist)
except OSError as e: try:
raise CosimulationError(_error.OSError, str(e)) sp = subprocess.Popen(exe, env=env, close_fds=False,
else: preexec_fn=close_rt_wf)
os.close(wt) except OSError as e:
os.close(rf) raise CosimulationError(_error.OSError, str(e))
while 1:
s = to_str(os.read(rt, _MAXLINE)) self._child = sp
if not s:
raise CosimulationError(_error.SimulationEnd) os.close(wt)
e = s.split() os.close(rf)
if e[0] == "FROM": while 1:
if int(e[1]) != 0: s = to_str(os.read(rt, _MAXLINE))
raise CosimulationError(_error.TimeZero, "$from_myhdl") if not s:
for i in range(2, len(e)-1, 2): raise CosimulationError(_error.SimulationEnd)
n = e[i] e = s.split()
if n in fromSignames: if e[0] == "FROM":
raise CosimulationError(_error.DuplicateSigNames, n) if int(e[1]) != 0:
if not n in kwargs: raise CosimulationError(_error.TimeZero, "$from_myhdl")
raise CosimulationError(_error.SigNotFound, n) for i in range(2, len(e)-1, 2):
fromSignames.append(n) n = e[i]
fromSigs.append(kwargs[n]) if n in fromSignames:
fromSizes.append(int(e[i+1])) raise CosimulationError(_error.DuplicateSigNames, n)
os.write(wf, b"OK") if not n in kwargs:
elif e[0] == "TO": raise CosimulationError(_error.SigNotFound, n)
if int(e[1]) != 0: fromSignames.append(n)
raise CosimulationError(_error.TimeZero, "$to_myhdl") fromSigs.append(kwargs[n])
for i in range(2, len(e)-1, 2): fromSizes.append(int(e[i+1]))
n = e[i] os.write(wf, b"OK")
if n in toSignames: elif e[0] == "TO":
raise CosimulationError(_error.DuplicateSigNames, n) if int(e[1]) != 0:
if not n in kwargs: raise CosimulationError(_error.TimeZero, "$to_myhdl")
raise CosimulationError(_error.SigNotFound, n) for i in range(2, len(e)-1, 2):
toSignames.append(n) n = e[i]
toSigs.append(kwargs[n]) if n in toSignames:
toSigDict[n] = kwargs[n] raise CosimulationError(_error.DuplicateSigNames, n)
toSizes.append(int(e[i+1])) if not n in kwargs:
os.write(wf, b"OK") raise CosimulationError(_error.SigNotFound, n)
elif e[0] == "START": toSignames.append(n)
if not toSignames: toSigs.append(kwargs[n])
raise CosimulationError(_error.NoCommunication) toSigDict[n] = kwargs[n]
os.write(wf, b"OK") toSizes.append(int(e[i+1]))
break os.write(wf, b"OK")
else: elif e[0] == "START":
raise CosimulationError("Unexpected cosim input") if not toSignames:
raise CosimulationError(_error.NoCommunication)
os.write(wf, b"OK")
break
else:
raise CosimulationError("Unexpected cosim input")
def _get(self): def _get(self):
if not self._getMode: if not self._getMode:

2
myhdl/_Simulation.py
View File

@ -78,7 +78,7 @@ class Simulation(object):
_simulator._cosim = 0 _simulator._cosim = 0
os.close(cosim._rt) os.close(cosim._rt)
os.close(cosim._wf) os.close(cosim._wf)
os.waitpid(cosim._child_pid, 0) cosim._child.wait()
if _simulator._tracing: if _simulator._tracing:
_simulator._tracing = 0 _simulator._tracing = 0
_simulator._tf.close() _simulator._tf.close()

2
myhdl/test/core/test_Cosimulation.py
View File

@ -60,7 +60,7 @@ class CosimulationTest(TestCase):
try: try:
Cosimulation("bla -x 45") Cosimulation("bla -x 45")
except CosimulationError as e: except CosimulationError as e:
self.assertTrue(e.kind in(_error.OSError, _error.SimulationEnd)) self.assertEqual(e.kind, _error.OSError)
else: else:
self.fail() self.fail()