Repository: robotika/osgar

Issue:
LogIndexedReader crash on growing files
```
  File "/home/zbynek/osgar-ws/indexedlidar/osgar/logger.py", line 203, in _create_index
    micros, channel, size = struct.unpack('IHH', data[pos:pos+8])
struct.error: unpack requires a buffer of 8 bytes
```

Discussion:

Patch:
diff --git a/osgar/logger.py b/osgar/logger.py
index c49ee76d..932a49a5 100644
--- a/osgar/logger.py
+++ b/osgar/logger.py
@@ -192,27 +192,40 @@ class LogAsserter(LogReader):
 
 
 def _create_index(data, pos, start_time=datetime.timedelta()):
+    """index is list of (pos,timestamp) where the last item
+       `pos` points just past the last valid packet
+       `timestamp` is last valid timestamp
+       this marker says where and when the next parsing
+       should continue when growing the file
+    """
     index = []
     end = len(data)
     times, prev = divmod(start_time, datetime.timedelta(microseconds=TIMESTAMP_OVERFLOW_STEP))
     us_offset = times * TIMESTAMP_OVERFLOW_STEP
     prev_micros = prev.microseconds + prev.seconds * 10**6 + prev.days * 24 * 3600 * 10**6
-    while pos + 8 < end:
+    dt = start_time
+    while pos + 8 <= end:
         start = pos
-        while True:
-            micros, channel, size = struct.unpack('IHH', data[pos:pos+8])
-            pos += 8 + size
-            if size < 0xFFFF:
-                break
+        header = data[pos:pos+8]
+        micros, channel, size = struct.unpack('IHH', header)
         if prev_micros > micros:
             us_offset += TIMESTAMP_OVERFLOW_STEP
         prev_micros = micros
         dt = datetime.timedelta(microseconds=micros+us_offset)
-        index.append( (start, dt) )
-    if pos == end: # complete file
-        index.append( (pos, dt ) )
+        pos += 8 + size
+        while size == 0xFFFF and pos + 8 <= end:
+            header = data[pos:pos+8]
+            micros_, channel, size = struct.unpack('IHH', header)
+            assert micros == micros_
+            pos += 8 + size
+        if pos > end: # current packet is not complete
+            index.append((start, dt))
+            return index
+        index.append((start, dt))
+    index.append((pos, dt))
     return index
 
+
 class LogIndexedReader:
     def __init__(self, filepath):
         self.filepath = filepath
@@ -231,7 +244,7 @@ class LogIndexedReader:
         os.close(self.fd)
 
     def __getitem__(self, index):
-        if index >= len(self):
+        if abs(index) >= len(self):
             raise IndexError("log index {} out of range".format(index))
         assert index < len(self)
         a = self.data[self.index[index][0]:self.index[index+1][0]]
@@ -247,13 +260,15 @@ class LogIndexedReader:
         dt = self.index[index][1]
         return dt, channel, data
 
-    def __len__(self):
+    def grow(self):
         if (len(self.data) < self.data.size()):
             self.data.close()
             self.data = mmap.mmap(self.fd, 0, access=mmap.ACCESS_READ)
-            append = _create_index(self.data, self.index[-1][0], self.index[-1][1])
-            self.index[-1:] = append
+            index = _create_index(self.data, self.index[-1][0], self.index[-1][1])
+            self.index[-1:] = index
+        return len(self.index)-1
 
+    def __len__(self):
         return len(self.index)-1
 
 


Test patch:
diff --git a/osgar/test_logger.py b/osgar/test_logger.py
index 47abd8ef..0aebd7b6 100644
--- a/osgar/test_logger.py
+++ b/osgar/test_logger.py
@@ -386,7 +386,7 @@ class LoggerIndexedTest(unittest.TestCase):
     #            dt, channel, data = next(log)
     #            self.assertEqual(dt, timedelta(hours=4))
 
-    def _test_large_blocks_with_time_overflow(self):
+    def test_large_blocks_with_time_overflow(self):
         with ExitStack() as at_exit:
             with patch('osgar.logger.datetime.datetime'):
                 osgar.logger.datetime.datetime = TimeStandsStill(datetime(2019, 1, 1))
@@ -400,7 +400,7 @@ class LoggerIndexedTest(unittest.TestCase):
                     osgar.logger.datetime.datetime = TimeStandsStill(datetime(2019, 1, 1, 2))
                     t3 = log.write(1, b'\x03'*100000)
                     self.assertEqual(t3, timedelta(hours=2))
-            with LogIndexedReader(filename) as log:
+            with LogIndexedReader(log.filename) as log:
                 dt, channel, data = log[1]
                 self.assertEqual(dt, timedelta(hours=0))
                 dt, channel, data = log[2]
@@ -433,6 +433,7 @@ class LoggerIndexedTest(unittest.TestCase):
 
                 f_out.write(f_in.read(100))
                 f_out.flush()
+                log.grow()
                 self.assertEqual(len(log), 2)
                 dt, channel, data = log[1]
                 self.assertEqual(data, b'\x01'*100)
@@ -443,6 +444,7 @@ class LoggerIndexedTest(unittest.TestCase):
                 f_out.write(f_in.read())
                 f_out.flush()
 
+                log.grow()
                 dt, channel, data = log[2]
                 self.assertEqual(dt, timedelta(hours=2))
                 self.assertEqual(data, b'\x02'*100)
@@ -454,4 +456,61 @@ class LoggerIndexedTest(unittest.TestCase):
         os.remove(partial)
         os.remove(filename)
 
+    def test_large_blocks_with_growing_file(self):
+        block_size = 100000 # fits into 2 packets, so 16 bytes overhead
+        with ExitStack() as at_exit:
+            with patch('osgar.logger.datetime.datetime'):
+                osgar.logger.datetime.datetime = TimeStandsStill(datetime(2019, 1, 1))
+                with osgar.logger.LogWriter(prefix='tmpA', note='') as log:
+                    at_exit.callback(os.remove, log.filename)
+                    t1 = log.write(1, b'\x01'*block_size)
+                    self.assertEqual(t1, timedelta(0))
+                    osgar.logger.datetime.datetime = TimeStandsStill(datetime(2019, 1, 1, 1))
+                    t2 = log.write(1, b'\x02'*block_size)
+                    self.assertEqual(t2, timedelta(hours=1))
+                    osgar.logger.datetime.datetime = TimeStandsStill(datetime(2019, 1, 1, 2))
+                    t3 = log.write(1, b'\x03'*block_size)
+                    self.assertEqual(t3, timedelta(hours=2))
+
+            partial = log.filename + '.part'
+            with open(log.filename, 'rb') as f_in, open(partial, 'wb') as f_out:
+                at_exit.callback(os.remove, partial)
+                # log file starts with 16 byte header
+                f_out.write(f_in.read(16))
+                f_out.write(f_in.read(100))
+                f_out.flush()
+
+                with LogIndexedReader(partial) as log:
+                    with self.assertRaises(IndexError):
+                        log[0]
+                    # add whole block plus 2 headers minus 100 already there
+                    f_out.write(f_in.read(block_size-100+16))
+                    f_out.flush()
+                    log.grow()
+                    self.assertEqual(len(log), 1)
+
+                    # add a partial header only
+                    f_out.write(f_in.read(4))
+                    f_out.flush()
+                    log.grow()
+                    self.assertEqual(len(log), 1)
+
+                    # add another block
+                    f_out.write(f_in.read(block_size+16))
+                    f_out.flush()
+                    log.grow()
+
+                    dt, channel, data = log[0]
+                    self.assertEqual(dt, timedelta(hours=0))
+                    dt, channel, data = log[1]
+                    self.assertEqual(dt, timedelta(hours=1))
+
+                    f_out.write(f_in.read())
+                    f_out.flush()
+                    log.grow()
+
+                    dt, channel, data = log[2]
+                    self.assertEqual(dt, timedelta(hours=2))
+
+
 # vim: expandtab sw=4 ts=4