Merge pull request #487 from mmeisinger/fix1787_extsize

OOIION-1787: Support performance enhancements

pyon/core/interceptor/encode.py

@@ -1,5 +1,7 @@
 #!/usr/bin/env python
 
+"""Messaging object encoder/decoder for IonObjects and numpy data"""
+
 import msgpack
 import sys
 import numpy as np
@@ -27,7 +29,8 @@ class EncodeTypes(object):
     NPVAL = 'n'
 
 
-# Global lazy load reference to the Pyon object registry (we be set on first use, not on load)
+# Global lazy load reference to the Pyon object registry (we be set on first use, not on load).
+# Note: We need this here so that the decode_ion/encode_ion functions can be imported (i.e. be static).
 obj_registry = None
 
 
@@ -37,14 +40,19 @@ def decode_ion(obj):
 
     # NOTE: Just matching on dict with "type_" is a bit weak
     if "type_" in obj:
+        if "__noion__" in obj:
+            # INTERNAL: Allow dicts to mask as IonObject without being decoded (with all defaults set and validated)
+            obj.pop("__noion__")
+            return obj
+
         global obj_registry
         if obj_registry is None:
             obj_registry = get_obj_registry()
 
         ion_obj = obj_registry.new(obj["type_"])
         for k, v in obj.iteritems():
             # unicode translate to utf8
-            # Note: That's not recursive within dicts/list or any other types
+            # Note: This is not recursive within dicts/list or any other types
             if isinstance(v, unicode):
                 v = v.encode('utf8')
             if k != "type_":
@@ -80,6 +88,7 @@ def decode_ion(obj):
 
     return obj
 
+
 def encode_ion(obj):
     """
     msgpack object hook to encode granule/numpy types and IonObjects.

pyon/core/interceptor/validate.py

@@ -1,5 +1,9 @@
+#!/usr/bin/env python
+
+"""Messaging interceptor to validate IonObjects"""
+
 from pyon.core.interceptor.interceptor import Interceptor
-from pyon.core.bootstrap import IonObject
+from pyon.core.bootstrap import IonObject, CFG
 from pyon.core.exception import BadRequest
 from pyon.core.object import IonObjectBase, walk
 from pyon.core.registry import is_ion_object
@@ -15,17 +19,16 @@ class ValidateInterceptor(Interceptor):
     def configure(self, config):
         if "enabled" in config:
             self.enabled = config["enabled"]
+        self.enabled = self.enabled and CFG.get_safe("container.objects.validate.interceptor", True)
         log.debug("ValidateInterceptor enabled: %s" % str(self.enabled))
 
     def outgoing(self, invocation):
         # Set validate flag in header if IonObject(s) found in message
-        #log.debug("ValidateInterceptor.outgoing: %s", invocation)
 
         #Nothing to validate on the outbound side
         return invocation
 
     def incoming(self, invocation):
-        #log.debug("ValidateInterceptor.incoming: %s", invocation)
 
         if self.enabled:
             payload = invocation.message

pyon/datastore/postgresql/README.txt

@@ -39,8 +39,6 @@ FUTURE FEATURES:
 
 OPEN QUESTIONS:
 - EXPLAIN ANALYZE queries
-- Use postgres 9.3 json operators?
-- Rewrite some of the json functions
 - list_datastores() must maintain compatible result, prefixing database with sysname (because of framework code)
-- Use msgpack for object serialization instead of json and custom accessor functions?
+- Use msgpack and BYTEA for object serialization instead of json and custom accessor functions?
 - rewrite association queries as select * from resources where id in (select from assoc)

pyon/datastore/postgresql/base_store.py

@@ -7,7 +7,9 @@
 import getpass
 import os.path
 from uuid import uuid4
-import simplejson as json
+# Note: standard json is faster than simplejson for dumps
+# See https://confluence.oceanobservatories.org/display/CIDev/Container+Messaging+Performance
+import json
 
 try:
     import psycopg2

pyon/net/channel.py

@@ -561,7 +561,7 @@ def _destroy_queue(self):
         assert self._recv_name# and isinstance(self._recv_name, tuple) and self._recv_name[1]
         assert self._transport
 
-        log.info("Destroying queue: %s", self._recv_name)
+        log.debug("Destroying queue: %s", self._recv_name)
         with self._ensure_transport():
             self._transport.delete_queue_impl(queue=self._recv_name.queue)