Remove upper limit on requests library version

We do this by creating a new requests.Session object for each request,
sending a "Connection: close" request header, and then explicitly
marking the connection for close after the response is read.

This is to avoid a longstanding race condition with HTTP keepalive
and server timeouts.  Due to data processing, capture, etc, requests
may be separated by an arbitrary delay.  If this delay is shorter
than the server's KeepAliveTimeout, the same connection is used.
If the delay is longer, a new connection is used.  If the delay is
the same, however, the request may be sent on the old connection at
the exact same time that the server closes it.  Typically, the
client sees the connection as closing between the request and the
response, which leads to "httplib.BadStatusLine" errors.

This patch avoids the race condition entirely by not using persistent
connections.

Another solution may be to detect those errors and retry the
connection, resending the request.  However, the race condition could
potentially show up in other places, like a closed connection during
the request body, not after.  Such an error could also be a legitimate
network condition or problem.  This solution should be more reliable,
and the overhead of each new connection will hopefully be minimal for
typical workloads.

nilmdb/client/httpclient.py

@@ -9,7 +9,7 @@ import requests
 
 class HTTPClient(object):
     """Class to manage and perform HTTP requests from the client"""
-    def __init__(self, baseurl = "", post_json = False):
+    def __init__(self, baseurl = "", post_json = False, verify_ssl = True):
         """If baseurl is supplied, all other functions that take
         a URL can be given a relative URL instead."""
         # Verify / clean up URL
@@ -18,9 +18,8 @@ class HTTPClient(object):
             reparsed = urlparse.urlparse("http://" + baseurl).geturl()
         self.baseurl = reparsed.rstrip('/') + '/'
 
-        # Build Requests session object, enable SSL verification
-        self.session = requests.Session()
-        self.session.verify = True
+        # Note whether we want SSL verification
+        self.verify_ssl = verify_ssl
 
         # Saved response, so that tests can verify a few things.
         self._last_response = {}
@@ -58,16 +57,34 @@ class HTTPClient(object):
                 raise Error(**args)
 
     def close(self):
-        self.session.close()
+        pass
 
     def _do_req(self, method, url, query_data, body_data, stream, headers):
         url = urlparse.urljoin(self.baseurl, url)
         try:
-            response = self.session.request(method, url,
-                                            params = query_data,
-                                            data = body_data,
-                                            stream = stream,
-                                            headers = headers)
+            # Create a new session, ensure we send "Connection: close",
+            # and explicitly close connection after the transfer.
+            # This is to avoid HTTP/1.1 persistent connections
+            # (keepalive), because they have fundamental race
+            # conditions when there are delays between requests:
+            # a new request may be sent at the same instant that the
+            # server decides to timeout the connection.
+            session = requests.Session()
+            if headers is None:
+                headers = {}
+            headers["Connection"] = "close"
+            response = session.request(method, url,
+                                       params = query_data,
+                                       data = body_data,
+                                       stream = stream,
+                                       headers = headers,
+                                       verify = self.verify_ssl)
+
+            # Close the connection.  If it's a generator (stream =
+            # True), the requests library shouldn't actually close the
+            # HTTP connection until all data has been read from the
+            # response.
+            session.close()
         except requests.RequestException as e:
             raise ServerError(status = "502 Error", url = url,
                               message = str(e.message))

nilmdb/cmdline/cmdline.py

@@ -121,7 +121,7 @@ class Cmdline(object):
         group = self.parser.add_argument_group("General options")
         group.add_argument("-h", "--help", action='help',
                            help='show this help message and exit')
-        group.add_argument("-V", "--version", action="version",
+        group.add_argument("-v", "--version", action="version",
                            version = nilmdb.__version__)
 
         group = self.parser.add_argument_group("Server")

nilmdb/cmdline/list.py

@@ -45,6 +45,8 @@ def setup(self, sub):
                        help="Show raw timestamps when printing times")
     group.add_argument("-l", "--layout", action="store_true",
                        help="Show layout type next to path name")
+    group.add_argument("-n", "--no-decim", action="store_true",
+                       help="Skip paths containing \"~decim-\"")
 
     return cmd
 
@@ -71,6 +73,8 @@ def cmd_list(self):
             (path, layout, int_min, int_max, rows, time) = stream[:6]
             if not fnmatch.fnmatch(path, argpath):
                 continue
+            if self.args.no_decim and "~decim-" in path:
+                continue
 
             if self.args.layout:
                 printf("%s %s\n", path, layout)

nilmdb/fsck/fsck.py

@@ -59,6 +59,8 @@ def retry_if_raised(exc, message = None, max_retries = 100):
 
 class Progress(object):
     def __init__(self, maxval):
+        if maxval == 0:
+            maxval = 1
         self.bar = progressbar.ProgressBar(
             maxval = maxval,
             widgets = [ progressbar.Percentage(), ' ',
@@ -381,7 +383,7 @@ class Fsck(object):
         err("*** Deleting the entire interval from SQL.\n")
         err("This may leave stale data on disk.  To fix that, copy all\n")
         err("data from this stream to a new stream, then remove all data\n")
-        err("from and destroy %s.\n")
+        err("from and destroy %s.\n", path)
         with self.sql:
             cur = self.sql.cursor()
             cur.execute("DELETE FROM ranges WHERE "

nilmdb/scripts/nilmdb_fsck.py

@@ -10,9 +10,8 @@ def main():
 
     parser = argparse.ArgumentParser(
         description = 'Check database consistency',
-        formatter_class = argparse.ArgumentDefaultsHelpFormatter)
-    parser.add_argument("-V", "--version", action="version",
-                        version = nilmdb.__version__)
+        formatter_class = argparse.ArgumentDefaultsHelpFormatter,
+        version = nilmdb.__version__)
     parser.add_argument("-f", "--fix", action="store_true",
                         default=False, help = 'Fix errors when possible '
                         '(which may involve removing data)')

nilmdb/scripts/nilmdb_server.py

@@ -10,10 +10,8 @@ def main():
 
     parser = argparse.ArgumentParser(
         description = 'Run the NilmDB server',
-        formatter_class = argparse.ArgumentDefaultsHelpFormatter)
-
-    parser.add_argument("-V", "--version", action="version",
-                        version = nilmdb.__version__)
+        formatter_class = argparse.ArgumentDefaultsHelpFormatter,
+        version = nilmdb.__version__)
 
     group = parser.add_argument_group("Standard options")
     group.add_argument('-a', '--address',

nilmdb/server/bulkdata.py

@@ -194,6 +194,9 @@ class BulkData(object):
         if oldospath == newospath:
             raise ValueError("old and new paths are the same")
 
+        # Remove Table object at old path from cache
+        self.getnode.cache_remove(self, oldunicodepath)
+
         # Move the table to a temporary location
         tmpdir = tempfile.mkdtemp(prefix = "rename-", dir = self.root)
         tmppath = os.path.join(tmpdir, "table")

nilmdb/server/nilmdb.py

@@ -83,8 +83,11 @@ _sql_schema_updates = {
 class NilmDB(object):
     verbose = 0
 
-    def __init__(self, basepath, max_results=None,
-                 max_removals=None, bulkdata_args=None):
+    def __init__(self, basepath,
+                 max_results=None,
+                 max_removals=None,
+                 max_int_removals=None,
+                 bulkdata_args=None):
         """Initialize NilmDB at the given basepath.
         Other arguments are for debugging / testing:
 
@@ -92,7 +95,10 @@ class NilmDB(object):
         stream_intervals or stream_extract response.
 
         'max_removals' is the max rows to delete at once
-        in stream_move.
+        in stream_remove.
+
+        'max_int_removals' is the max intervals to delete
+        at once in stream_remove.
 
         'bulkdata_args' is kwargs for the bulkdata module.
         """
@@ -134,6 +140,9 @@ class NilmDB(object):
         # Remove up to this many rows per call to stream_remove.
         self.max_removals = max_removals or 1048576
 
+        # Remove up to this many intervals per call to stream_remove.
+        self.max_int_removals = max_int_removals or 4096
+
     def get_basepath(self):
         return self.basepath
 
@@ -643,13 +652,22 @@ class NilmDB(object):
         to_remove = Interval(start, end)
         removed = 0
         remaining = self.max_removals
+        int_remaining = self.max_int_removals
         restart = None
 
         # Can't remove intervals from within the iterator, so we need to
         # remember what's currently in the intersection now.
         all_candidates = list(intervals.intersection(to_remove, orig = True))
 
+        remove_start = None
+        remove_end = None
+
         for (dbint, orig) in all_candidates:
+            # Stop if we've hit the max number of interval removals
+            if int_remaining <= 0:
+                restart = dbint.start
+                break
+
             # Find row start and end
             row_start = self._find_start(table, dbint)
             row_end = self._find_end(table, dbint)
@@ -670,14 +688,29 @@ class NilmDB(object):
             # Remove interval from the database
             self._remove_interval(stream_id, orig, dbint)
 
-            # Remove data from the underlying table storage
-            table.remove(row_start, row_end)
+            # Remove data from the underlying table storage,
+            # coalescing adjacent removals to reduce the number of calls
+            # to table.remove.
+            if remove_end == row_start:
+                # Extend our coalesced region
+                remove_end = row_end
+            else:
+                # Perform previous removal, then save this one
+                if remove_end is not None:
+                    table.remove(remove_start, remove_end)
+                remove_start = row_start
+                remove_end = row_end
 
             # Count how many were removed
             removed += row_end - row_start
             remaining -= row_end - row_start
+            int_remaining -= 1
 
             if restart is not None:
                 break
 
+        # Perform any final coalesced removal
+        if remove_end is not None:
+            table.remove(remove_start, remove_end)
+
         return (removed, restart)

nilmdb/utils/mustclose.py

@@ -28,10 +28,13 @@ def must_close(errorfile = sys.stderr, wrap_verify = False):
 
         @wrap_class_method
         def __del__(orig, self, *args, **kwargs):
-            if "_must_close" in self.__dict__:
-                fprintf(errorfile, "error: %s.close() wasn't called!\n",
-                        self.__class__.__name__)
-            return orig(self, *args, **kwargs)
+            try:
+                if "_must_close" in self.__dict__:
+                    fprintf(errorfile, "error: %s.close() wasn't called!\n",
+                            self.__class__.__name__)
+                return orig(self, *args, **kwargs)
+            except: # pragma: no cover
+                pass
 
         @wrap_class_method
         def close(orig, self, *args, **kwargs):

nilmdb/utils/serializer.py

@@ -117,7 +117,10 @@ def serializer_proxy(obj_or_type):
             return ret
 
         def __del__(self):
-            self.__call_queue.put((None, None, None, None))
-            self.__thread.join()
+            try:
+                self.__call_queue.put((None, None, None, None))
+                self.__thread.join()
+            except: # pragma: no cover
+                pass
 
     return SerializerObjectProxy(obj_or_type)

setup.py

@@ -117,7 +117,7 @@ setup(name='nilmdb',
                            'python-dateutil',
                            'pytz',
                            'psutil >= 0.3.0',
-                           'requests >= 1.1.0, < 2.0.0',
+                           'requests >= 1.1.0',
                            'progressbar >= 2.2',
                            ],
       packages = [ 'nilmdb',

tests/test_client.py

@@ -690,40 +690,15 @@ class TestClient(object):
         client.close()
 
     def test_client_12_persistent(self):
-        # Check that connections are persistent when they should be.
-        # This is pretty hard to test; we have to poke deep into
-        # the Requests library.
+        # Check that connections are NOT persistent.  Rather than trying
+        # to verify this at the TCP level, just make sure that the response
+        # contained a "Connection: close" header.
         with nilmdb.client.Client(url = testurl) as c:
-            def connections():
-                try:
-                    poolmanager = c.http._last_response.connection.poolmanager
-                    pool = poolmanager.pools[('http','localhost',32180)]
-                    return (pool.num_connections, pool.num_requests)
-                except Exception:
-                    raise SkipTest("can't get connection info")
-
-            # First request makes a connection
             c.stream_create("/persist/test", "uint16_1")
-            eq_(connections(), (1, 1))
+            eq_(c.http._last_response.headers["Connection"], "close")
 
-            # Non-generator
-            c.stream_list("/persist/test")
-            eq_(connections(), (1, 2))
-            c.stream_list("/persist/test")
-            eq_(connections(), (1, 3))
-
-            # Generators
-            for x in c.stream_intervals("/persist/test"):
-                pass
-            eq_(connections(), (1, 4))
-            for x in c.stream_intervals("/persist/test"):
-                pass
-            eq_(connections(), (1, 5))
-
-            # Clean up
-            c.stream_remove("/persist/test")
             c.stream_destroy("/persist/test")
-            eq_(connections(), (1, 7))
+            eq_(c.http._last_response.headers["Connection"], "close")
 
     def test_client_13_timestamp_rounding(self):
         # Test potentially bad timestamps (due to floating point

tests/test_cmdline.py

@@ -21,13 +21,17 @@ from testutil.helpers import *
 
 testdb = "tests/cmdline-testdb"
 
-def server_start(max_results = None, max_removals = None, bulkdata_args = {}):
+def server_start(max_results = None,
+                 max_removals = None,
+                 max_int_removals = None,
+                 bulkdata_args = {}):
     global test_server, test_db
     # Start web app on a custom port
     test_db = nilmdb.utils.serializer_proxy(nilmdb.server.NilmDB)(
         testdb,
         max_results = max_results,
         max_removals = max_removals,
+        max_int_removals = max_int_removals,
         bulkdata_args = bulkdata_args)
     test_server = nilmdb.server.Server(test_db, host = "127.0.0.1",
                                        port = 32180, stoppable = False,
@@ -290,6 +294,7 @@ class TestCmdline(object):
         self.ok("create /newton/zzz/rawnotch uint16_9")
         self.ok("create /newton/prep float32_8")
         self.ok("create /newton/raw uint16_6")
+        self.ok("create /newton/raw~decim-1234 uint16_6")
 
         # Create a stream that already exists
         self.fail("create /newton/raw uint16_6")
@@ -305,13 +310,23 @@ class TestCmdline(object):
         self.fail("create /newton/zzz float32_8")
         self.contain("subdirs of this path already exist")
 
-        # Verify we got those 3 streams and they're returned in
+        # Verify we got those 4 streams and they're returned in
         # alphabetical order.
         self.ok("list -l")
         self.match("/newton/prep float32_8\n"
                    "/newton/raw uint16_6\n"
+                   "/newton/raw~decim-1234 uint16_6\n"
                    "/newton/zzz/rawnotch uint16_9\n")
 
+        # No decimated streams if -n specified
+        self.ok("list -n -l")
+        self.match("/newton/prep float32_8\n"
+                   "/newton/raw uint16_6\n"
+                   "/newton/zzz/rawnotch uint16_9\n")
+
+        # Delete that decimated stream
+        self.ok("destroy /newton/raw~decim-1234")
+
         # Match just one type or one path.  Also check
         # that --path is optional
         self.ok("list --layout /newton/raw")
@@ -869,14 +884,26 @@ class TestCmdline(object):
         self.ok("destroy -R /newton/prep") # destroy again
 
     def test_14_remove_files(self):
-        # Test BulkData's ability to remove when data is split into
-        # multiple files.  Should be a fairly comprehensive test of
-        # remove functionality.
-        # Also limit max_removals, to cover more functionality.
+        # Limit max_removals, to cover more functionality.
         server_stop()
         server_start(max_removals = 4321,
                      bulkdata_args = { "file_size" : 920, # 23 rows per file
                                        "files_per_dir" : 3 })
+        self.do_remove_files()
+        self.ok("destroy -R /newton/prep") # destroy again
+
+    def test_14b_remove_files_maxint(self):
+        # Limit max_int_removals, to cover more functionality.
+        server_stop()
+        server_start(max_int_removals = 1,
+                     bulkdata_args = { "file_size" : 920, # 23 rows per file
+                                       "files_per_dir" : 3 })
+        self.do_remove_files()
+
+    def do_remove_files(self):
+        # Test BulkData's ability to remove when data is split into
+        # multiple files.  Should be a fairly comprehensive test of
+        # remove functionality.
 
         # Insert data.  Just for fun, insert out of order
         self.ok("create /newton/prep float32_8")