Skip to content

build.{plat,res}: post-lib.io cleanup. #1310

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Apr 11, 2024
Merged

build.{plat,res}: post-lib.io cleanup. #1310

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
139 changes: 9 additions & 130 deletions amaranth/build/plat.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,34 +142,15 @@ def prepare(self, elaboratable, name="top", **kwargs):
def missing_domain_error(name):
raise RuntimeError("Missing domain in pin fragment")

def add_pin_fragment(pin, pin_fragment):
pin_fragment = Fragment.get(pin_fragment, self)
pin_fragment._propagate_domains(missing_domain_error)
pin_fragment = DomainLowerer()(pin_fragment)
fragment.add_subfragment(pin_fragment, name=f"pin_{pin.name}")

for pin, port, attrs, invert in self.iter_single_ended_pins():
if pin.dir == "i":
add_pin_fragment(pin, self.get_input(pin, port, attrs, invert))
if pin.dir == "o":
add_pin_fragment(pin, self.get_output(pin, port, attrs, invert))
if pin.dir == "oe":
add_pin_fragment(pin, self.get_tristate(pin, port, attrs, invert))
if pin.dir == "io":
add_pin_fragment(pin, self.get_input_output(pin, port, attrs, invert))

for pin, port, attrs, invert in self.iter_differential_pins():
if pin.dir == "i":
add_pin_fragment(pin, self.get_diff_input(pin, port, attrs, invert))
if pin.dir == "o":
add_pin_fragment(pin, self.get_diff_output(pin, port, attrs, invert))
if pin.dir == "oe":
add_pin_fragment(pin, self.get_diff_tristate(pin, port, attrs, invert))
if pin.dir == "io":
add_pin_fragment(pin, self.get_diff_input_output(pin, port, attrs, invert))

fragment = Design(fragment, [], hierarchy=(name,))
return self.toolchain_prepare(fragment, name, **kwargs)
for pin, port, buffer in self.iter_pins():
buffer = Fragment.get(buffer, self)
buffer._propagate_domains(missing_domain_error)
buffer = DomainLowerer()(buffer)
fragment.add_subfragment(buffer, name=f"pin_{pin.name}")

ports = [(port.name, port, None) for port in self.iter_ports()]
design = Design(fragment, ports, hierarchy=(name,))
return self.toolchain_prepare(design, name, **kwargs)

@abstractmethod
def toolchain_prepare(self, fragment, name, **kwargs):
Expand All @@ -186,108 +167,6 @@ def toolchain_program(self, products, name, **kwargs):
raise NotImplementedError("Platform '{}' does not support programming"
.format(type(self).__name__))

def _check_feature(self, feature, pin, attrs, valid_xdrs, valid_attrs):
if len(valid_xdrs) == 0:
raise NotImplementedError("Platform '{}' does not support {}"
.format(type(self).__name__, feature))
elif pin.xdr not in valid_xdrs:
raise NotImplementedError("Platform '{}' does not support {} for XDR {}"
.format(type(self).__name__, feature, pin.xdr))

if not valid_attrs and attrs:
raise NotImplementedError("Platform '{}' does not support attributes for {}"
.format(type(self).__name__, feature))

@staticmethod
def _invert_if(invert, value):
if invert:
return ~value
else:
return value

def get_input(self, pin, port, attrs, invert):
self._check_feature("input", pin, attrs,
valid_xdrs=(0, 1, 2), valid_attrs=True)

m = Module()
if pin.xdr == 0:
m.submodules.buf = buf = io.Buffer(io.Direction.Input, port)
m.d.comb += pin.i.eq(buf.i)
elif pin.xdr == 1:
m.domains.input = cd_input = ClockDomain(reset_less=True)
m.submodules.buf = buf = io.FFBuffer(io.Direction.Input, port, i_domain="input")
m.d.comb += pin.i.eq(buf.i)
m.d.comb += cd_input.clk.eq(pin.i_clk)
elif pin.xdr == 2:
m.domains.input = cd_input = ClockDomain(reset_less=True)
m.submodules.buf = buf = io.DDRBuffer(io.Direction.Input, port, i_domain="input")
m.d.comb += pin.i0.eq(buf.i[0])
m.d.comb += pin.i1.eq(buf.i[1])
m.d.comb += cd_input.clk.eq(pin.i_clk)
return m

def get_output(self, pin, port, attrs, invert):
self._check_feature("output", pin, attrs,
valid_xdrs=(0, 1, 2), valid_attrs=True)

m = Module()
if pin.xdr == 0:
m.submodules.buf = buf = io.Buffer(io.Direction.Output, port)
m.d.comb += buf.o.eq(pin.o)
elif pin.xdr == 1:
m.domains.output = cd_output = ClockDomain(reset_less=True)
m.submodules.buf = buf = io.FFBuffer(io.Direction.Output, port, o_domain="output")
m.d.comb += buf.o.eq(pin.o)
m.d.comb += cd_output.clk.eq(pin.o_clk)
elif pin.xdr == 2:
m.domains.output = cd_output = ClockDomain(reset_less=True)
m.submodules.buf = buf = io.DDRBuffer(io.Direction.Output, port, o_domain="output")
m.d.comb += buf.o[0].eq(pin.o0)
m.d.comb += buf.o[1].eq(pin.o1)
m.d.comb += cd_output.clk.eq(pin.o_clk)
if pin.dir == "oe":
m.d.comb += buf.oe.eq(pin.oe)
return m

get_tristate = get_output

def get_input_output(self, pin, port, attrs, invert):
self._check_feature("single-ended input/output", pin, attrs,
valid_xdrs=(0, 1, 2), valid_attrs=True)

m = Module()
if pin.xdr == 0:
m.submodules.buf = buf = io.Buffer(io.Direction.Bidir, port)
m.d.comb += pin.i.eq(buf.i)
m.d.comb += buf.o.eq(pin.o)
m.d.comb += buf.oe.eq(pin.oe)
elif pin.xdr == 1:
m.domains.input = cd_input = ClockDomain(reset_less=True)
m.domains.output = cd_output = ClockDomain(reset_less=True)
m.submodules.buf = buf = io.FFBuffer(io.Direction.Bidir, port, i_domain="input", o_domain="output")
m.d.comb += pin.i.eq(buf.i)
m.d.comb += buf.o.eq(pin.o)
m.d.comb += buf.oe.eq(pin.oe)
m.d.comb += cd_input.clk.eq(pin.i_clk)
m.d.comb += cd_output.clk.eq(pin.o_clk)
elif pin.xdr == 2:
m.domains.input = cd_input = ClockDomain(reset_less=True)
m.domains.output = cd_output = ClockDomain(reset_less=True)
m.submodules.buf = buf = io.DDRBuffer(io.Direction.Bidir, port, i_domain="input", o_domain="output")
m.d.comb += pin.i0.eq(buf.i[0])
m.d.comb += pin.i1.eq(buf.i[1])
m.d.comb += buf.o[0].eq(pin.o0)
m.d.comb += buf.o[1].eq(pin.o1)
m.d.comb += buf.oe.eq(pin.oe)
m.d.comb += cd_input.clk.eq(pin.i_clk)
m.d.comb += cd_output.clk.eq(pin.o_clk)
return m

get_diff_input = get_input
get_diff_output = get_output
get_diff_tristate = get_tristate
get_diff_input_output = get_input_output


class TemplatedPlatform(Platform):
toolchain = property(abstractmethod(lambda: None))
Expand Down
168 changes: 104 additions & 64 deletions amaranth/build/res.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,8 @@
from collections import OrderedDict

from ..hdl._ast import *
from ..lib.io import *
from ..lib import wiring
from ..hdl import *
from ..hdl._ast import SignalDict
from ..lib import wiring, io

from .dsl import *

Expand All @@ -24,6 +24,78 @@ def __init__(self, name, attrs):
self.attrs = attrs


class PinBuffer(Elaboratable):
def __init__(self, pin, port):
if pin.xdr not in (0, 1, 2):
raise ValueError(f"Unsupported 'xdr' value {pin.xdr}")
self.pin = pin
self.port = port

def elaborate(self, platform):
m = Module()

if self.pin.dir == "i":
if self.pin.xdr == 0:
m.submodules.buf = buf = io.Buffer(io.Direction.Input, self.port)
m.d.comb += self.pin.i.eq(buf.i)
elif self.pin.xdr == 1:
m.domains.input = cd_input = ClockDomain(reset_less=True)
m.submodules.buf = buf = io.FFBuffer(io.Direction.Input, self.port, i_domain="input")
m.d.comb += self.pin.i.eq(buf.i)
m.d.comb += cd_input.clk.eq(self.pin.i_clk)
elif self.pin.xdr == 2:
m.domains.input = cd_input = ClockDomain(reset_less=True)
m.submodules.buf = buf = io.DDRBuffer(io.Direction.Input, self.port, i_domain="input")
m.d.comb += self.pin.i0.eq(buf.i[0])
m.d.comb += self.pin.i1.eq(buf.i[1])
m.d.comb += cd_input.clk.eq(self.pin.i_clk)
if self.pin.dir in ("o", "oe"):
if self.pin.xdr == 0:
m.submodules.buf = buf = io.Buffer(io.Direction.Output, self.port)
m.d.comb += buf.o.eq(self.pin.o)
elif self.pin.xdr == 1:
m.domains.output = cd_output = ClockDomain(reset_less=True)
m.submodules.buf = buf = io.FFBuffer(io.Direction.Output, self.port, o_domain="output")
m.d.comb += buf.o.eq(self.pin.o)
m.d.comb += cd_output.clk.eq(self.pin.o_clk)
elif self.pin.xdr == 2:
m.domains.output = cd_output = ClockDomain(reset_less=True)
m.submodules.buf = buf = io.DDRBuffer(io.Direction.Output, self.port, o_domain="output")
m.d.comb += buf.o[0].eq(self.pin.o0)
m.d.comb += buf.o[1].eq(self.pin.o1)
m.d.comb += cd_output.clk.eq(self.pin.o_clk)
if self.pin.dir == "oe":
m.d.comb += buf.oe.eq(self.pin.oe)
if self.pin.dir == "io":
if self.pin.xdr == 0:
m.submodules.buf = buf = io.Buffer(io.Direction.Bidir, self.port)
m.d.comb += self.pin.i.eq(buf.i)
m.d.comb += buf.o.eq(self.pin.o)
m.d.comb += buf.oe.eq(self.pin.oe)
elif self.pin.xdr == 1:
m.domains.input = cd_input = ClockDomain(reset_less=True)
m.domains.output = cd_output = ClockDomain(reset_less=True)
m.submodules.buf = buf = io.FFBuffer(io.Direction.Bidir, self.port, i_domain="input", o_domain="output")
m.d.comb += self.pin.i.eq(buf.i)
m.d.comb += buf.o.eq(self.pin.o)
m.d.comb += buf.oe.eq(self.pin.oe)
m.d.comb += cd_input.clk.eq(self.pin.i_clk)
m.d.comb += cd_output.clk.eq(self.pin.o_clk)
elif self.pin.xdr == 2:
m.domains.input = cd_input = ClockDomain(reset_less=True)
m.domains.output = cd_output = ClockDomain(reset_less=True)
m.submodules.buf = buf = io.DDRBuffer(io.Direction.Bidir, self.port, i_domain="input", o_domain="output")
m.d.comb += self.pin.i0.eq(buf.i[0])
m.d.comb += self.pin.i1.eq(buf.i[1])
m.d.comb += buf.o[0].eq(self.pin.o0)
m.d.comb += buf.o[1].eq(self.pin.o1)
m.d.comb += buf.oe.eq(self.pin.oe)
m.d.comb += cd_input.clk.eq(self.pin.i_clk)
m.d.comb += cd_output.clk.eq(self.pin.o_clk)

return m


class ResourceManager:
def __init__(self, resources, connectors):
self.resources = OrderedDict()
Expand All @@ -33,8 +105,11 @@ def __init__(self, resources, connectors):
self.connectors = OrderedDict()
self._conn_pins = OrderedDict()

# Constraint lists
# List of all IOPort instances created
self._ports = []
# List of (pin, port, buffer) pairs for non-dir="-" requests.
self._pins = []
# Constraint list
self._clocks = SignalDict()

self.add_resources(resources)
Expand Down Expand Up @@ -139,11 +214,12 @@ def resolve(resource, dir, xdr, path, attrs):
direction = phys.dir
if isinstance(phys, Pins):
phys_names = phys.map_names(self._conn_pins, resource)
io = IOPort(len(phys), name="__".join(path) + "__io", metadata=[
iop = IOPort(len(phys), name="__".join(path) + "__io", metadata=[
PortMetadata(name, attrs)
for name in phys_names
])
port = SingleEndedPort(io, invert=phys.invert, direction=direction)
self._ports.append(iop)
port = io.SingleEndedPort(iop, invert=phys.invert, direction=direction)
if isinstance(phys, DiffPairs):
phys_names_p = phys.p.map_names(self._conn_pins, resource)
phys_names_n = phys.n.map_names(self._conn_pins, resource)
Expand All @@ -156,11 +232,8 @@ def resolve(resource, dir, xdr, path, attrs):
PortMetadata(name, attrs)
for name in phys_names_n
])
port = DifferentialPort(p, n, invert=phys.invert, direction=direction)
if dir == "-":
pin = None
else:
pin = wiring.flipped(Pin(len(phys), dir, xdr=xdr, path=path))
self._ports += [p, n]
port = io.DifferentialPort(p, n, invert=phys.invert, direction=direction)

for phys_name in phys_names:
if phys_name in self._phys_reqd:
Expand All @@ -171,12 +244,16 @@ def resolve(resource, dir, xdr, path, attrs):
".".join(self._phys_reqd[phys_name])))
self._phys_reqd[phys_name] = path

self._ports.append((resource, pin, port, attrs))

if pin is not None and resource.clock is not None:
self.add_clock_constraint(pin.i, resource.clock.frequency)
if dir == "-":
return port
else:
pin = wiring.flipped(io.Pin(len(phys), dir, xdr=xdr, path=path))
buffer = PinBuffer(pin, port)
self._pins.append((pin, port, buffer))

return pin if pin is not None else port
if resource.clock is not None:
self.add_clock_constraint(pin.i, resource.clock.frequency)
return pin

else:
assert False # :nocov:
Expand All @@ -188,56 +265,19 @@ def resolve(resource, dir, xdr, path, attrs):
self._requested[resource.name, resource.number] = value
return value

def iter_single_ended_pins(self):
for res, pin, port, attrs in self._ports:
if pin is None:
continue
if isinstance(res.ios[0], Pins):
yield pin, port, attrs, res.ios[0].invert

def iter_differential_pins(self):
for res, pin, port, attrs in self._ports:
if pin is None:
continue
if isinstance(res.ios[0], DiffPairs):
yield pin, port, attrs, res.ios[0].invert

def should_skip_port_component(self, port, attrs, component):
return False
def iter_pins(self):
yield from self._pins

def iter_ports(self):
for res, pin, port, attrs in self._ports:
if isinstance(res.ios[0], Pins):
if not self.should_skip_port_component(port, attrs, "io"):
yield port.io
elif isinstance(res.ios[0], DiffPairs):
if not self.should_skip_port_component(port, attrs, "p"):
yield port.p
if not self.should_skip_port_component(port, attrs, "n"):
yield port.n
else:
assert False

def iter_port_constraints(self):
for res, pin, port, attrs in self._ports:
if isinstance(res.ios[0], Pins):
if not self.should_skip_port_component(port, attrs, "io"):
yield port.io.name, res.ios[0].map_names(self._conn_pins, res), attrs
elif isinstance(res.ios[0], DiffPairs):
if not self.should_skip_port_component(port, attrs, "p"):
yield port.p.name, res.ios[0].p.map_names(self._conn_pins, res), attrs
if not self.should_skip_port_component(port, attrs, "n"):
yield port.n.name, res.ios[0].n.map_names(self._conn_pins, res), attrs
else:
assert False
yield from self._ports

def iter_port_constraints_bits(self):
for port_name, pin_names, attrs in self.iter_port_constraints():
if len(pin_names) == 1:
yield port_name, pin_names[0], attrs
for port in self._ports:
if len(port) == 1:
yield port.name, port.metadata[0].name, port.metadata[0].attrs
else:
for bit, pin_name in enumerate(pin_names):
yield f"{port_name}[{bit}]", pin_name, attrs
for bit, meta in enumerate(port.metadata):
yield f"{port.name}[{bit}]", meta.name, meta.attrs

def add_clock_constraint(self, clock, frequency):
if isinstance(clock, ClockSignal):
Expand Down Expand Up @@ -267,11 +307,11 @@ def iter_clock_constraints(self):
# Constraints on nets with no corresponding input pin (e.g. PLL or SERDES outputs) are not
# affected.
pin_i_to_port = SignalDict()
for res, pin, port, attrs in self._ports:
for pin, port, _fragment in self._pins:
if hasattr(pin, "i"):
if isinstance(res.ios[0], Pins):
if isinstance(port, io.SingleEndedPort):
pin_i_to_port[pin.i] = port.io
elif isinstance(res.ios[0], DiffPairs):
elif isinstance(port, io.DifferentialPort):
pin_i_to_port[pin.i] = port.p
else:
assert False
Expand Down
Loading