Merge lp:~jderose/filestore/fadvise into lp:filestore

[Jason Gerard DeRose]

In the future we should experiment with POSIX_FADV_WILLNEED, but for now, this simple change provides an impressive performance improvement (almost 10%).

Changes:

 * When needed, monkey patch `os` module with dummy `os.posix_fadvise()`

 * reader() now calls on.posix_fadvise() with os.POSIX_FADV_SEQUENTIAL before starting read

 * batch_reader() now calls on.posix_fadvise() with os.POSIX_FADV_SEQUENTIAL before starting each read

 * batch_import_iter() now uses a queue size of 8 instead of 16 (meaning now at most 64 MiB will be in-flight in the queue, instead of 128 MiB)

 * reader() and batch_reader() now log.exception() when an exception is caught to make debugging easier

 * benchmark.py script now uses batch_import_iter(), other misc changes

[lp:~jderose/filestore/fadvise]

275. By Jason Gerard DeRose on 2012-10-30

Wow, with os.POSIX_FADV_WILLNEED, now I'm up to almost an 18% improvement, 31.2 MB/s

276. By Jason Gerard DeRose on 2012-10-30

Changed reader() to use os.POSIX_FADV_WILLNEED also

[Jason Gerard DeRose (jderose)]

Okay, I decided to undergo said experiment right now... and holy crap, fast.

Using os.POSIX_FADV_WILLNEED, I'm up to almost an 18% performance improvement.

31.2 MB/s over USB2, up from 26.5 MB/s.

There are a few things that need further tuning on the Dmedia side of things, but the filestore read performance is now pretty stellar (under Python3.3 anyway).

[James Raymond (jamesmr)]

review: Approve

=== modified file 'benchmark.py'
--- benchmark.py	2011-09-07 03:46:52 +0000
+++ benchmark.py	2012-10-30 02:40:25 +0000
@@ -23,6 +23,10 @@
 
 """
 Benchmarking utilities for in-tree tests (not installed).
+
+Clear page cache like this:
+
+    sync ; echo 3 | sudo tee /proc/sys/vm/drop_caches
 """
 
 import sys
@@ -30,30 +34,32 @@
 import time
 import math
 
-
 import filestore
 
 
-UNITS_BASE10 = (
+BYTES10 = (
     'bytes',
     'kB',
     'MB',
     'GB',
     'TB',
     'PB',
+    'EB',
+    'ZB',
+    'YB',
 )
 
-def units_base10(size):
+def bytes10(size):
     """
     Return *size* bytes to 3 significant digits in SI base-10 units.
 
     For example:
 
-    >>> units_base10(1000)
+    >>> bytes10(1000)
     '1 kB'
-    >>> units_base10(29481537)
+    >>> bytes10(29481537)
     '29.5 MB'
-    >>> units_base10(392012353)
+    >>> bytes10(392012353)
     '392 MB'
 
     For additional details, see:
@@ -63,17 +69,15 @@
     if size is None:
         return None
     if size < 0:
-        raise ValueError('size must be greater than zero; got %r' % size)
-    if size >= 10 ** 18:
-        raise ValueError('size must be smaller than 10**18; got %r' % size)
+        raise ValueError('size must be >= 0; got {!r}'.format(size))
     if size == 0:
         return '0 bytes'
     if size == 1:
         return '1 byte'
-    i = int(math.floor(math.log(size, 1000)))
-    s = (size / (1000.0 ** i) if i > 0 else size)
+    i = min(int(math.floor(math.log(size, 1000))), len(BYTES10) - 1)
+    s = (size / (1000 ** i) if i > 0 else size)
     return (
-        '%.*g %s' % (3, s, UNITS_BASE10[i])
+        '{:.3g} {}'.format(s, BYTES10[i])
     )
 
 
@@ -88,7 +92,7 @@
 
     def rate(self, total_size):
         per_sec = total_size / self.elapsed
-        return '{} per second'.format(units_base10(per_sec))
+        return '{} per second'.format(bytes10(per_sec))
 
 
 if __name__ == '__main__':
@@ -100,18 +104,17 @@
         raise SystemExit(
             'ERROR: source is not a directory: {!r}'.format(d)
         )
+    if len(args) > 1:
+        dst = filestore.FileStore(path.abspath(args[1]))
+    else:
+        dst = filestore.TempFileStore()
 
     print('Benchmarking {!r}'.format(d))
     batch = filestore.scandir(d)
     print('  {} files'.format(batch.count))
-    print('  total size: {}'.format(units_base10(batch.size)))
-
+    print('  total size: {}'.format(bytes10(batch.size)))
     t = Timer()
-    for f in batch.files:
-        src_fp = open(f.name, 'rb')
-        while True:
-            data = src_fp.read(filestore.LEAF_SIZE)
-            if not data:
-                break
-    print('Finished in {}'.format(t.done()))
+    for (file, ch) in filestore.batch_import_iter(batch, dst):
+        pass
+    print('Finished in {:.3f}'.format(t.done()))
     print('Read at {}'.format(t.rate(batch.size)))

=== modified file 'filestore.py'
--- filestore.py	2012-10-19 20:01:09 +0000
+++ filestore.py	2012-10-30 02:40:25 +0000
@@ -140,6 +140,16 @@
     https://launchpad.net/dmedia
 """
 
+# Monkey patch python3.2 with dummy os.posix_fadvise():
+if not hasattr(os, 'posix_fadvise'):
+    def _posix_fadvise(fd, offset, length, advice):
+        pass
+    os.posix_fadvise = _posix_fadvise
+    os.POSIX_FADV_SEQUENTIAL = 2
+    os.POSIX_FADV_WILLNEED = 3
+    os.POSIX_FADV_DONTNEED = 4
+
+
 
 ##############################
 # The Dmedia Hashing Protocol:
@@ -679,15 +689,31 @@
                 TYPE_ERROR.format('src_fp', IO_READABLE, type(src_fp), src_fp)
             )
         src_fp.seek(0)
+        fd = src_fp.fileno()
+        st = os.fstat(fd)
+        if st.st_size > 0:
+            os.posix_fadvise(fd, 0, st.st_size, os.POSIX_FADV_SEQUENTIAL)
         index = 0
         while True:
             data = src_fp.read(LEAF_SIZE)
             if not data:
                 queue.put(None)
                 break
+            if len(data) == LEAF_SIZE:
+                os.posix_fadvise(fd,
+                    (index + 1) * LEAF_SIZE,
+                    LEAF_SIZE,
+                    os.POSIX_FADV_WILLNEED
+                )
+            os.posix_fadvise(fd,
+                index * LEAF_SIZE,
+                len(data),
+                os.POSIX_FADV_DONTNEED
+            )
             queue.put(Leaf(index, data))
             index += 1
     except Exception as e:
+        log.exception('Error reading %r', src_fp.name)
         queue.put(e)
 
 
@@ -740,18 +766,33 @@
     try:
         for file in batch.files:
             queue.put(file)
-            log.info('Opening file %r', file.name)
+            log.info('Opening %r', file)
             src_fp = open(file.name, 'rb')
+            fd = src_fp.fileno()
+            if file.size > 0:
+                os.posix_fadvise(fd, 0, file.size, os.POSIX_FADV_SEQUENTIAL)
             index = 0
             while True:
                 data = src_fp.read(LEAF_SIZE)
                 if not data:
                     queue.put(EndFile)
                     break
+                if len(data) == LEAF_SIZE:
+                    os.posix_fadvise(fd,
+                        (index + 1) * LEAF_SIZE,
+                        LEAF_SIZE,
+                        os.POSIX_FADV_WILLNEED
+                    )
+                os.posix_fadvise(fd,
+                    index * LEAF_SIZE,
+                    len(data),
+                    os.POSIX_FADV_DONTNEED
+                )
                 queue.put(Leaf(index, data))
                 index += 1
         queue.put(EndBatch)
     except Exception as e:
+        log.exception('batch.count=%r, batch.size=%r', batch.count, batch.size)
         queue.put(e)
 
 
@@ -802,7 +843,7 @@
     if callback is not None:
         count = 0
         size = 0
-    q = SmartQueue(16)
+    q = SmartQueue(8)
     thread = _start_thread(batch_reader, batch, q)
     while True:
         file = q.get()