Merge pull request #448 from basho/features/lrb/rts-842-perf-test-t2b-encoding

READY: Add support for Erlang term-to-binary encoding

Author: lukebakken

.gitignore

@@ -1,22 +1,17 @@
+README.rst
 *.pyc
 .python-version
 __pycache__/
-
 .tox/
-
-.tox/
-
 docs/_build
-
 .*.swp
 .coverage
-
+riak-*/
 py-build/
 dist/
-
 riak.egg-info/
 *.egg
 .eggs/
-
 #*#
 *~
+*.ps1

MANIFEST.in

@@ -1,6 +1,7 @@
 include docs/*
 include riak/erl_src/*
 include README.md
+include README.rst
 include LICENSE
 include RELNOTES.md
 include version.py

buildbot/Makefile

@@ -26,7 +26,6 @@ compile:
 
 lint:
 	@pip install --upgrade pep8 flake8
-	@cd ..; pep8 --exclude=riak/pb riak *.py
 	@cd ..; flake8 --exclude=riak/pb riak *.py
 
 test: setup test_normal test_security
@@ -46,7 +45,7 @@ test_security:
 test_timeseries:
 	@echo "Testing Riak Python Client (timeseries)"
 	@$(RIAK_ADMIN) security disable
-	@RIAK_TEST_PROTOCOL='pbc' RUN_YZ=0 RUN_DATATYPES=0 RUN_INDEXES=1 RUN_TIMESERIES=1 ./tox_runner.sh ..
+	@RIAK_TEST_PROTOCOL='pbc' RIAK_TEST_PB_PORT=8087 RUN_YZ=0 RUN_DATATYPES=0 RUN_INDEXES=1 RUN_TIMESERIES=1 ./tox_runner.sh ..
 
 setup:
 	./tox_setup.sh

docs/advanced.rst

@@ -93,7 +93,7 @@ Transports
 
 .. currentmodule:: riak.transports.transport
 
-.. autoclass:: RiakTransport
+.. autoclass:: Transport
    :members:
    :private-members:
 
@@ -124,20 +124,24 @@ HTTP Transport
 
 .. currentmodule:: riak.transports.http
 
-.. autoclass:: RiakHttpPool
+.. autoclass:: HttpPool
 
 .. autofunction:: is_retryable
 
-.. autoclass:: RiakHttpTransport
+.. autoclass:: HttpTransport
    :members:
 
-^^^^^^^^^^^^^^^^^^^^^^^^^^
-Protocol Buffers Transport
-^^^^^^^^^^^^^^^^^^^^^^^^^^
+^^^^^^^^^^^^^
+TCP Transport
+^^^^^^^^^^^^^
+
+.. currentmodule:: riak.transports.tcp
 
-.. currentmodule:: riak.transports.pbc
+.. autoclass:: TcpPool
+
+.. autofunction:: is_retryable
 
-.. autoclass:: RiakPbcTransport
+.. autoclass:: TcpTransport
    :members:
 
 ---------

riak/benchmark.py

@@ -1,24 +1,9 @@
-"""
-Copyright 2013 Basho Technologies, Inc.
-
-This file is provided to you under the Apache License,
-Version 2.0 (the "License"); you may not use this file
-except in compliance with the License.  You may obtain
-a copy of the License at
-
-http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing,
-software distributed under the License is distributed on an
-"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-KIND, either express or implied.  See the License for the
-specific language governing permissions and limitations
-under the License.
-"""
-
 from __future__ import print_function
+
 import os
 import gc
+import sys
+import traceback
 
 __all__ = ['measure', 'measure_with_rehearsal']
 
@@ -172,5 +157,7 @@ def __exit__(self, exc_type, exc_val, exc_tb):
         elif exc_type is KeyboardInterrupt:
             return False
         else:
-            print("EXCEPTION! %r" % ((exc_type, exc_val, exc_tb),))
-        return True
+            msg = "EXCEPTION! type: %r val: %r" % (exc_type, exc_val)
+            print(msg, file=sys.stderr)
+            traceback.print_tb(exc_tb)
+        return True if exc_type is None else False

riak/benchmarks/timeseries.py

@@ -0,0 +1,74 @@
+import datetime
+import random
+import sys
+
+import riak.benchmark as benchmark
+
+from multiprocessing import cpu_count
+from riak import RiakClient
+
+# logger = logging.getLogger()
+# logger.level = logging.DEBUG
+# logger.addHandler(logging.StreamHandler(sys.stdout))
+
+# batch sizes 8, 16, 32, 64, 128, 256
+if len(sys.argv) != 3:
+    raise AssertionError(
+            'first arg is batch size, second arg is true / false'
+            'for use_ttb')
+
+rowcount = 32768
+batchsz = int(sys.argv[1])
+if rowcount % batchsz != 0:
+    raise AssertionError('rowcount must be divisible by batchsz')
+use_ttb = sys.argv[2].lower() == 'true'
+
+epoch = datetime.datetime.utcfromtimestamp(0)
+onesec = datetime.timedelta(0, 1)
+
+weather = ['typhoon', 'hurricane', 'rain', 'wind', 'snow']
+rows = []
+for i in range(rowcount):
+    ts = datetime.datetime(2016, 1, 1, 12, 0, 0) + \
+        datetime.timedelta(seconds=i)
+    family_idx = i % batchsz
+    series_idx = i % batchsz
+    family = 'hash{:d}'.format(family_idx)
+    series = 'user{:d}'.format(series_idx)
+    w = weather[i % len(weather)]
+    temp = (i % 100) + random.random()
+    row = [family, series, ts, w, temp]
+    key = [family, series, ts]
+    rows.append(row)
+
+print("Benchmarking timeseries:")
+print("   Use TTB: {}".format(use_ttb))
+print("Batch Size: {}".format(batchsz))
+print("      CPUs: {}".format(cpu_count()))
+print("      Rows: {}".format(len(rows)))
+print()
+
+tbl = 'GeoCheckin'
+h = 'riak-test'
+n = [
+    {'host': h, 'pb_port': 10017},
+    {'host': h, 'pb_port': 10027},
+    {'host': h, 'pb_port': 10037},
+    {'host': h, 'pb_port': 10047},
+    {'host': h, 'pb_port': 10057}
+]
+client = RiakClient(nodes=n, protocol='pbc',
+                    transport_options={'use_ttb': use_ttb})
+table = client.table(tbl)
+
+with benchmark.measure() as b:
+    for i in (1, 2, 3):
+        with b.report('populate-%d' % i):
+            for i in range(0, rowcount, batchsz):
+                x = i
+                y = i + batchsz
+                r = rows[x:y]
+                ts_obj = table.new(r)
+                result = ts_obj.store()
+                if result is not True:
+                    raise AssertionError("expected success")

riak/client/__init__.py

@@ -11,8 +11,8 @@
 from riak.mapreduce import RiakMapReduceChain
 from riak.resolver import default_resolver
 from riak.table import Table
-from riak.transports.http import RiakHttpPool
-from riak.transports.pbc import RiakPbcPool
+from riak.transports.http import HttpPool
+from riak.transports.tcp import TcpPool
 from riak.security import SecurityCreds
 from riak.util import lazy_property, bytes_to_str, str_to_bytes
 from six import string_types, PY2
@@ -68,7 +68,7 @@ class RiakClient(RiakMapReduceChain, RiakClientOperations):
     PROTOCOLS = ['http', 'pbc']
 
     def __init__(self, protocol='pbc', transport_options={}, nodes=None,
-                 credentials=None, multiget_pool_size=None, **unused_args):
+                 credentials=None, multiget_pool_size=None, **kwargs):
         """
         Construct a new ``RiakClient`` object.
 
@@ -88,19 +88,19 @@ def __init__(self, protocol='pbc', transport_options={}, nodes=None,
            CPUs in the system
         :type multiget_pool_size: int
         """
-        unused_args = unused_args.copy()
+        kwargs = kwargs.copy()
 
         if nodes is None:
-            self.nodes = [self._create_node(unused_args), ]
+            self.nodes = [self._create_node(kwargs), ]
         else:
             self.nodes = [self._create_node(n) for n in nodes]
 
         self._multiget_pool_size = multiget_pool_size
         self.protocol = protocol or 'pbc'
         self._resolver = None
         self._credentials = self._create_credentials(credentials)
-        self._http_pool = RiakHttpPool(self, **transport_options)
-        self._pb_pool = RiakPbcPool(self, **transport_options)
+        self._http_pool = HttpPool(self, **transport_options)
+        self._tcp_pool = TcpPool(self, **transport_options)
 
         if PY2:
             self._encoders = {'application/json': default_encoder,
@@ -167,7 +167,7 @@ def _get_client_id(self):
     def _set_client_id(self, client_id):
         for http in self._http_pool:
             http.client_id = client_id
-        for pb in self._pb_pool:
+        for pb in self._tcp_pool:
             pb.client_id = client_id
 
     client_id = property(_get_client_id, _set_client_id,
@@ -298,8 +298,8 @@ def close(self):
         """
         if self._http_pool is not None:
             self._http_pool.clear()
-        if self._pb_pool is not None:
-            self._pb_pool.clear()
+        if self._tcp_pool is not None:
+            self._tcp_pool.clear()
 
     def _create_node(self, n):
         if isinstance(n, RiakNode):

riak/client/multiget.py

@@ -1,26 +1,9 @@
-"""
-Copyright 2013 Basho Technologies, Inc.
-
-This file is provided to you under the Apache License,
-Version 2.0 (the "License"); you may not use this file
-except in compliance with the License.  You may obtain
-a copy of the License at
-
-http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing,
-software distributed under the License is distributed on an
-"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-KIND, either express or implied.  See the License for the
-specific language governing permissions and limitations
-under the License.
-"""
-
 from __future__ import print_function
 from collections import namedtuple
 from threading import Thread, Lock, Event
 from multiprocessing import cpu_count
 from six import PY2
+
 if PY2:
     from Queue import Queue
 else:
@@ -177,8 +160,8 @@ def multiget(client, keys, **options):
         :meth:`RiakBucket.get <riak.bucket.RiakBucket.get>`
     :type options: dict
     :rtype: list
-
     """
+
     outq = Queue()
 
     if 'pool' in options:

riak/client/transport.py

@@ -1,23 +1,6 @@
-"""
-Copyright 2012 Basho Technologies, Inc.
-
-This file is provided to you under the Apache License,
-Version 2.0 (the "License"); you may not use this file
-except in compliance with the License.  You may obtain
-a copy of the License at
-
-http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing,
-software distributed under the License is distributed on an
-"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-KIND, either express or implied.  See the License for the
-specific language governing permissions and limitations
-under the License.
-"""
 from contextlib import contextmanager
 from riak.transports.pool import BadResource
-from riak.transports.pbc import is_retryable as is_pbc_retryable
+from riak.transports.tcp import is_retryable as is_pbc_retryable
 from riak.transports.http import is_retryable as is_http_retryable
 import threading
 from six import PY2
@@ -49,7 +32,7 @@ class RiakClientTransport(object):
     # These will be set or redefined by the RiakClient initializer
     protocol = 'pbc'
     _http_pool = None
-    _pb_pool = None
+    _tcp_pool = None
     _locals = _client_locals()
 
     def _get_retry_count(self):
@@ -163,8 +146,8 @@ def _choose_pool(self, protocol=None):
             protocol = self.protocol
         if protocol == 'http':
             pool = self._http_pool
-        elif protocol == 'pbc':
-            pool = self._pb_pool
+        elif protocol == 'tcp' or protocol == 'pbc':
+            pool = self._tcp_pool
         else:
             raise ValueError("invalid protocol %s" % protocol)
         return pool

riak/codecs/__init__.py

@@ -0,0 +1,25 @@
+import collections
+
+from riak import RiakError
+
+Msg = collections.namedtuple('Msg',
+                             ['msg_code', 'data', 'resp_code'],
+                             verbose=False)
+
+
+class Codec(object):
+    def parse_msg(self):
+        raise NotImplementedError('parse_msg not implemented')
+
+    def maybe_incorrect_code(self, resp_code, expect=None):
+        if expect and resp_code != expect:
+            raise RiakError("unexpected message code: %d, expected %d"
+                            % (resp_code, expect))
+
+    def maybe_riak_error(self, err_code, msg_code, data=None):
+        if msg_code == err_code:
+            if data is None:
+                raise RiakError('no error provided!')
+            return data
+        else:
+            return None