""" pl_quiesce.py -- quiesce named AXI DMA masters in the currently-loaded PL design before downloading a new bitstream. Rationale --------- When PYNQ downloads a bitstream, any AXI master still actively transacting on an HP port belongs to the *previous* design. Reconfiguration erases the logic but not the outstanding transaction, which can leave the new design's first memory-side access wedged. Nothing in the new design's RTL can prevent this -- the offending logic no longer exists by the time the new design does -- so the fix has to happen on the PS side, before download. Scope ----- Only DMA instances whose names the caller supplies are touched. Everything else in the loaded design is left strictly alone, including the ~9 axi_dma instances in the stock base.bit that PYNQ loads at boot. This matters: an MMIO read to an IP whose clock is not running never returns, hanging the PS rather than raising, and there is no way to probe for that safely -- the probe is the read. Usage ----- from pl_quiesce import quiesce_pl DMA_NAMES = ["AXI_DMA_SG", "AXI_DMA_NOSG", "axi_dma_1", "axi_dma_2"] quiesce_pl(DMA_NAMES) To find the DMA names in a design, load it and run: [n for n, ip in gen.ip_dict.items() if 'axi_dma' in str(ip.get('type',''))] What the code does: 1. Ask the PL server for ip_dict — the metadata for the design loaded right now, i.e. the previous one, since you haven't downloaded yet. If unavailable, print and return. 2. Filter it to IPs that are both named in names (leaf name, so hierarchies still match) and have axi_dma in their type field. 3. Print matched names. If nothing matched, return — that's the normal path on the first download after boot. Then for each matched DMA, map its registers from phys_addr, and run these steps on MM2S, then again on S2MM: 4. Read DMACR and check bit 0, Run/Stop. Record it as was_running. 5. If it's set, clear it. Per PG021 the engine finishes its current transaction and then asserts Halted — this is the graceful stop, and it's what lets an in-flight memory transaction complete rather than being abandoned mid-burst. 6. Poll DMASR bit 0 for up to 500 ms waiting for Halted. Record whether it got there. 7. Set DMACR bit 2, the soft reset, whether or not step 6 succeeded. It's the harder stop if the graceful one timed out, and free if it didn't. 8. Poll for that bit to self-clear, up to 500 ms. Record it. 9. Print the per-channel line and append the result dict. Every step that can fail is caught and recorded rather than raised, so a partial teardown still lets the download proceed — with a visible INCOMPLETE in the log. """ import time from pynq import MMIO, PL # AXI DMA register map (PG021) MM2S_DMACR = 0x00 MM2S_DMASR = 0x04 S2MM_DMACR = 0x30 S2MM_DMASR = 0x34 DMACR_RS = 1 << 0 # Run/Stop DMACR_RESET = 1 << 2 # Soft reset, self-clearing DMASR_HALTED = 1 << 0 # 1 = halted def _quiesce_channel(mmio, cr_off, sr_off, label, timeout_s=0.5, verbose=True): """Halt one DMA channel, then soft-reset it. Never raises.""" result = {"channel": label, "was_running": None, "halted": None, "reset_cleared": None, "note": None} try: cr = mmio.read(cr_off) except Exception as e: result["note"] = f"unreadable ({e.__class__.__name__})" return result result["was_running"] = bool(cr & DMACR_RS) if result["was_running"]: # Clear Run/Stop. Per PG021 the engine halts after the current # transaction completes; DMASR.Halted asserts when it is done. try: mmio.write(cr_off, cr & ~DMACR_RS) except Exception as e: result["note"] = f"halt write failed ({e.__class__.__name__})" return result deadline = time.time() + timeout_s while time.time() < deadline: if mmio.read(sr_off) & DMASR_HALTED: break time.sleep(0.001) result["halted"] = bool(mmio.read(sr_off) & DMASR_HALTED) else: result["halted"] = True result["note"] = "was already stopped" # Soft reset unconditionally: the stronger hammer if the graceful halt timed # out, and free if it succeeded. try: mmio.write(cr_off, mmio.read(cr_off) | DMACR_RESET) except Exception as e: result["note"] = f"reset write failed ({e.__class__.__name__})" return result deadline = time.time() + timeout_s while time.time() < deadline: if not (mmio.read(cr_off) & DMACR_RESET): break time.sleep(0.001) result["reset_cleared"] = not (mmio.read(cr_off) & DMACR_RESET) if verbose: ok = result["halted"] and result["reset_cleared"] extra = f" [{result['note']}]" if result["note"] else "" print(f" {label:24s} was_running={result['was_running']} " f"halted={result['halted']} reset_cleared={result['reset_cleared']} " f" {'ok' if ok else 'INCOMPLETE'}{extra}") return result def quiesce_pl(names, verbose=True): """Halt + soft-reset the named AXI DMAs in the currently-loaded PL design. Parameters ---------- names : iterable of str DMA instance names to quiesce. Required -- there is no default, so the set of IPs this will touch is always visible at the call site. Matching is on the leaf name, so a DMA inside a block-design hierarchy ("hier_0/AXI_DMA_SG") still matches "AXI_DMA_SG". Names not present in the loaded design are skipped without complaint; on the first download after a boot, none of them will be. verbose : bool Print per-channel status. Returns ------- list of per-channel result dicts. Never raises: a teardown that can fail partway is worse than none, because it leaves a master live while letting the caller believe it is safe to proceed. """ wanted = set(names or ()) results = [] if not wanted: if verbose: print("quiesce_pl: no DMA names given; nothing to do") return results try: ip_dict = dict(PL.ip_dict) except Exception as e: if verbose: print(f"quiesce_pl: no PL metadata ({e}); skipping") return results # Require both the name match and the IP type: a name collision with a # non-DMA block would otherwise send DMA register offsets at something that # is not one. dmas = {n: ip for n, ip in ip_dict.items() if "axi_dma" in str(ip.get("type", "")) and n.split("/")[-1] in wanted} if verbose: try: loaded = PL.bitfile_name except Exception: loaded = "" print(f"quiesce_pl: matched {len(dmas)} of {len(wanted)} named DMA(s) " f"in {loaded}") if not dmas: return results for name, ip in dmas.items(): if verbose: print(f" {name} @ 0x{ip['phys_addr']:08x}") try: mmio = MMIO(ip["phys_addr"], ip.get("addr_range", 0x10000)) except Exception as e: if verbose: print(f" could not map ({e.__class__.__name__}); skipping") continue results.append(_quiesce_channel(mmio, MM2S_DMACR, MM2S_DMASR, f"{name}.MM2S", verbose=verbose)) results.append(_quiesce_channel(mmio, S2MM_DMACR, S2MM_DMASR, f"{name}.S2MM", verbose=verbose)) return results