flash driver checks prior sector erasure. Add littlefs_test.

bdevice.py

@@ -13,6 +13,8 @@
 # of data to arbitrary addresses. IOW .readwrite handles physical block structure
 # while ioctl supports a virtual block size.
 
+from micropython import const
+
 class BlockDevice:
 
     def __init__(self, nbits, nchips, chip_size):
@@ -87,6 +89,8 @@ class BlockDevice:
 # .flush(cache, addr)  Erase physical sector and write out an entire cached sector.
 # .readwrite As per base class.
 
+_RDBUFSIZE = const(32)  # Size of read buffer for erasure test
+
 class FlashDevice(BlockDevice):
 
     def __init__(self, nbits, nchips, chip_size, sec_size):
@@ -94,6 +98,8 @@ class FlashDevice(BlockDevice):
         self.sec_size = sec_size
         self._cache_mask = sec_size - 1  # For 4K sector size: 0xfff
         self._fmask = self._cache_mask ^ 0x3fffffff  # 4K -> 0x3ffff000
+        self._buf = bytearray(_RDBUFSIZE)
+        self._mvbuf = memoryview(self._buf)
         self._cache = bytearray(sec_size)  # Cache always contains one sector
         self._mvd = memoryview(self._cache)
         self._acache = 0  # Address in chip of byte 0 of current cached sector
@@ -151,3 +157,16 @@ class FlashDevice(BlockDevice):
 
     def initialise(self):
         self._fill_cache(0)
+
+    # Return True if a sector is erased. Assumes erased == ff.
+    def is_empty(self, addr):
+        mvb = self._mvbuf
+        erased = True
+        nbufs = self.sec_size // _RDBUFSIZE  # Read buffers per sector
+        for _ in range(nbufs):
+            self.rdchip(addr, mvb)
+            if any(True for x in mvb if x != 0xff):
+                erased = False
+                break
+            addr += _RDBUFSIZE
+        return erased

flash/FLASH.md

@@ -31,7 +31,7 @@ connected to 3V3 or left unconnected.
 
 | Flash | Signal  |  PB | Signal |
 |:-----:|:-------:|:---:|:------:|
-| 1     |  CS     | Y5  | SS/    |
+| 1     |  CS/    | Y5  | SS/    |
 | 2     |  SO     | Y7  | MISO   |
 | 3     |  WP/    | nc  | -      |
 | 4     |  Vss    | Gnd | Gnd    |
@@ -64,8 +64,11 @@ SPI bus is fast: wiring should be short and direct.
  1. `flash_spi.py` Device driver.
  2. `bdevice.py` (In root directory) Base class for the device driver.
  3. `flash_test.py` Test programs for above.
+ 4. `littlefs_test.py` Torture test for the littlefs filesystem on the flash
+ array.
 
-Installation: copy files 1 and 2 (optionally 3) to the target filesystem.
+Installation: copy files 1 and 2 (3 & 4 are optional) to the target filesystem.
+Test scripts assume two chips with CS/ pins wired to Pyboard pins Y4 and Y5.
 
 # 4. The device driver
 

flash/flash_spi.py

@@ -124,7 +124,7 @@ class FLASH(FlashDevice):
     def _wait_rdy(self):  # After a write, wait for device to become ready
         mvp = self._mvp
         cs = self._ccs  # Chip is already current
-        while True:
+        while True:  # TODO read status register 2, raise OSError on nonzero.
             mvp[0] = _RDSR1
             cs(0)
             self._spi.write_readinto(mvp[:2], mvp[:2])
@@ -150,6 +150,7 @@ class FLASH(FlashDevice):
 
     # Erase sector. Address is start byte address of sector.
     def _sector_erase(self, addr):
+        if not self.is_empty(addr):
             self._getaddr(addr, 1)
             cs = self._ccs  # Current chip select from _getaddr
             mvp = self._mvp

flash/flash_test.py

@@ -123,16 +123,35 @@ def cptest():
 
 
 # ***** TEST OF HARDWARE *****
-def full_test():
+def full_test(count=10):
-    eep = get_device()
+    flash = get_device()
-    page = 0
+    for n in range(count):
-    for sa in range(0, len(eep), 256):
         data = uos.urandom(256)
-        eep[sa:sa + 256] = data
+        while True:
-        got = eep[sa:sa + 256]
+            sa = int.from_bytes(uos.urandom(4), 'little') & 0x3fffffff
-        if got == data:
+            if sa < (flash._a_bytes - 256):
-            print('Page {} passed'.format(page))
-        else:
-            print('Page {} readback failed.'.format(page))
                 break
-        page += 1
+        flash[sa:sa + 256] = data
+        flash.synchronise()
+        got = flash[sa:sa + 256]
+        if got == data:
+            print('Pass {} address {:08x} passed'.format(n, sa))
+            if sa & 0xfff > (4096 -253):
+                print('cross boundary')
+        else:
+            print('Pass {} address {:08x} readback failed.'.format(n, sa))
+            sa1 = sa & 0xfff
+            print('Bounds {} to {}'.format(sa1, sa1+256))
+#            flash.synchronise()
+            got1 = flash[sa:sa + 256]
+            if got1 == data:
+                print('second attempt OK')
+            else:
+                print('second attempt fail', got == got1)
+                for n, g in enumerate(got):
+                    if g != data[n]:
+                        print('{} {:2x} {:2x} {:2x}'.format(n, data[n], g, got1[n]))
+            break
+
+# Fail seems to be on write: two readbacks are always identical.
+# Error is on MSB of byte 0: write a 1 and get 0

flash/littlefs_test.py

@@ -0,0 +1,84 @@
+# littlefs_test.py Extended filesystem test of flash devices
+# Create multiple binary files of varying length and verify that they can be
+# read back correctly. Rewrite files with new lengths then check that all files
+# are OK.
+
+import uos
+from machine import SPI, Pin
+from flash_spi import FLASH
+
+cspins = (Pin(Pin.board.Y5, Pin.OUT, value=1), Pin(Pin.board.Y4, Pin.OUT, value=1))
+
+# Return flash array. Adapt for platforms other than Pyboard.
+def get_device():
+    if uos.uname().machine.split(' ')[0][:4] == 'PYBD':
+        Pin.board.EN_3V3.value(1)
+    flash = FLASH(SPI(2, baudrate=5_000_000), cspins)
+    print('Instantiated Flash')
+    return flash
+
+directory = '/fl_ext'
+a = bytearray(range(256))
+b = bytearray(256)
+files = {}  # n:length
+errors = 0
+
+def fname(n):
+    return '{}/{:05d}'.format(directory, n + 1)  # Names start 00001
+
+def fcreate(n):  # Create a binary file of random length
+    length = int.from_bytes(uos.urandom(2), 'little') + 1  # 1-65536 bytes
+    linit = length
+    with open(fname(n), 'wb') as f:
+        while(length):
+            nw = min(length, 256)
+            f.write(a[:nw])
+            length -= nw
+    files[n] = length
+    return linit
+
+def fcheck(n):
+    length = files[n]
+    with open(fname(n), 'rb') as f:
+        while(length):
+            nr = f.readinto(b)
+            if not nr:
+                return False
+            if a[:nr] != b[:nr]:
+                return False
+            length -= nr
+    return True
+
+def check_all():
+    global errors
+    for n in files:
+        if fcheck(n):
+            print('File {:d} OK'.format(n))
+        else:
+            print('Error in file', n)
+            errors += 1
+    print('Total errors:', errors)
+
+
+def remove_all():
+    for n in files:
+        uos.remove(fname(n))
+
+def main():
+    eep = get_device()
+    try:
+        uos.mount(eep, directory)
+    except OSError:  # Already mounted
+        pass
+    for n in range(128):
+        length = fcreate(n)
+        print('Created', n, length)
+    print('Created files', files)
+    check_all()
+    for _ in range(100):
+        for x in range(5):  # Rewrite 5 files with new lengths
+            n = int.from_bytes(uos.urandom(1), 'little') & 0x7f
+            length = fcreate(n)
+            print('Rewrote', n, length)
+        check_all()
+    remove_all()