New subject: Change in vdsm[master]: stomp: Optimize stomp encoder\decoder

17 Nov 2014

Saggi Mizrahi has uploaded a new change for review.
Change subject: jsonrpc: Optimize stomp encoder\decoder
......................................................................
jsonrpc: Optimize stomp encoder\decoder
Change-Id: I341e46463e654ae0a086cd49af18aa336b86cb6e
Signed-off-by: Saggi Mizrahi smizrahi@redhat.com
---
M lib/yajsonrpc/stomp.py
1 file changed, 60 insertions(+), 42 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/55/35255/1

diff --git a/lib/yajsonrpc/stomp.py b/lib/yajsonrpc/stomp.py
index 884bff6..1dda6d7 100644
--- a/lib/yajsonrpc/stomp.py
+++ b/lib/yajsonrpc/stomp.py
@@ -14,8 +14,8 @@
 # Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
import logging
+import os
 import socket
-import cStringIO
 from threading import Timer, Event
 from uuid import uuid4
 from collections import deque
@@ -27,6 +27,8 @@
_RE_ESCAPE_SEQUENCE = re.compile(r"\(.)")
+_RE_ENCODE_CHARS = re.compile(r"[\r\n\:]")
+
 _EC_DECODE_MAP = {
     r"\": "\",
     r"r": "\r",
@@ -34,7 +36,12 @@
     r"c": ":",
 }
-_ESCAPE_CHARS = (('\\', '\'), ('\r', '\r'), ('\n', '\n'), ('\c', ':'))
+_EC_ENCODE_MAP = {
+    ":": "\c",
+    "\": "\\",
+    "\r": "\r",
+    "\n": "\n",
+}
class Command:
@@ -70,6 +77,8 @@
class Frame(object):
+    __slots__ = ("headers", "command", "body")
+
     def __init__(self, command="", headers=None, body=None):
         self.command = command
         if headers is None:
@@ -87,15 +96,19 @@
         if body is not None:
             self.headers["content-length"] = len(body)
-        data = self.command + '\n'
-        data += '\n'.join(["%s:%s" % (encodeValue(key), encodeValue(value))
-                          for key, value in self.headers.iteritems()])
-        data += '\n\n'
-        if body is not None:
-            data += body
+        data = [self.command, '\n']
+        for key, value in self.headers.iteritems():
+            data.append(encodeValue(key))
+            data.append(":")
+            data.append(encodeValue(value))
+            data.append("\n")
-        data += "\0"
-        return data
+        data.append('\n')
+        if body is not None:
+            data.append(body)
+
+        data.append("\0")
+        return ''.join(data)
def __repr__(self):
         return "<StompFrame command=%s>" % (repr(self.command))
@@ -111,24 +124,23 @@
         raise ValueError("Contains illigal charachter `:`")
try:
-        return _RE_ESCAPE_SEQUENCE.sub(
-            lambda m: _EC_DECODE_MAP[m.groups()[0]],
-            s
+        s = _RE_ESCAPE_SEQUENCE.sub(
+            lambda m: _EC_DECODE_MAP[m.group(0)],
+            s,
         )
     except KeyError as e:
         raise ValueError("Containes invalid escape squence `\%s`" % e.args[0])
+    return s.decode('utf-8')
+
def encodeValue(s):
-    if not isinstance(s, (str, unicode)):
-        s = str(s)
-    for escaped, raw in _ESCAPE_CHARS:
-        s = s.replace(raw, escaped)
-
     if isinstance(s, unicode):
         s = s.encode('utf-8')
+    elif not isinstance(s, str):
+        s = str(s)
-    return s
+    return _RE_ENCODE_CHARS.sub(lambda m: _EC_ENCODE_MAP[m.group(0)], s)
class Parser(object):
@@ -142,22 +154,29 @@
             self._STATE_HEADER: self._parse_header,
             self._STATE_BODY: self._parse_body}
         self._frames = deque()
-        self._state = self._STATE_CMD
+        self._change_state(self._STATE_CMD)
         self._contentLength = -1
         self._flush()
+    def _change_state(self, new_state):
+        self._state = new_state
+        self._state_cb = self._states[new_state]
+
     def _flush(self):
-        self._buffer = cStringIO.StringIO()
+        self._buffer = ""
+
+    def _write_buffer(self, buff):
+        self._buffer += buff
+
+    def _get_buffer(self):
+        return self._buffer
def _handle_terminator(self, term):
-        if term not in self._buffer.getvalue():
+        res, _, rest = self._buffer.partition(term)
+        if not _:
             return None
-        data = self._buffer.getvalue()
-        res, rest = data.split(term, 1)
-        self._flush()
-        self._buffer.write(rest)
-
+        self._buffer = rest
         return res
def _parse_command(self):
@@ -174,7 +193,7 @@
         cmd = decodeValue(cmd)
         self._tmpFrame = Frame(cmd)
-        self._state = self._STATE_HEADER
+        self._change_state(self._STATE_HEADER)
         return True
def _parse_header(self):
@@ -182,13 +201,13 @@
         if header is None:
             return False
-        headers = self._tmpFrame.headers
         if len(header) > 0 and header[-1] == '\r':
             header = header[:-1]
+        headers = self._tmpFrame.headers
         if header == "":
             self._contentLength = int(headers.get('content-length', -1))
-            self._state = self._STATE_BODY
+            self._change_state(self._STATE_BODY)
             return True
key, value = header.split(":", 1)
@@ -206,7 +225,7 @@
def _pushFrame(self):
         self._frames.append(self._tmpFrame)
-        self._state = self._STATE_CMD
+        self._change_state(self._STATE_CMD)
         self._tmpFrame = None
         self._contentLength = -1
@@ -226,16 +245,16 @@
         return True
def _parse_body_length(self):
-        buf = self._buffer
+        buf = self._get_buffer()
         cl = self._contentLength
-        ndata = buf.tell()
+        ndata = len(buf)
         if ndata < cl:
             return False
remainingBytes = 0
         self._flush()
-        body = buf.getvalue()
-        self._buffer.write(body[cl + 1:])
+        body = buf
+        self._write_buffer(body[cl + 1:])
         body = body[:cl]
if remainingBytes == 0:
@@ -249,9 +268,8 @@
         return len(self._frames)
def parse(self, data):
-        states = self._states
-        self._buffer.write(data)
-        while states[self._state]():
+        self._write_buffer(data)
+        while self._state_cb():
             pass
def popFrame(self):
@@ -347,7 +365,7 @@
def put_subscribe(self, destination, ack=None):
         subid = self._aclient.subscribe(self._adisp, destination, ack)
-        sub = Subsciption(self, subid, ack)
+        sub = Subscription(self, subid, ack)
         self._registerSubscription(sub)
         return sub
@@ -378,7 +396,7 @@
 class AsyncDispatcher(object):
     log = logging.getLogger("stomp.AsyncDispatcher")
-    def __init__(self, frameHandler, bufferSize=1024):
+    def __init__(self, frameHandler, bufferSize=4096):
         self._frameHandler = frameHandler
         self._bufferSize = bufferSize
         self._parser = Parser()
@@ -471,7 +489,7 @@
         return True
def _milis(self):
-        return int(round(time.time() * 1000))
+        return int(os.times()[4] * 1000)
class AsyncClient(object):
@@ -552,7 +570,7 @@
         return subscriptionID
-class Subsciption(object):
+class Subscription(object):
     def __init__(self, client, subid, ack):
         self._ack = ack
         self._subid = subid
-- 
To view, visit http://gerrit.ovirt.org/35255
To unsubscribe, visit http://gerrit.ovirt.org/settings

Gerrit-MessageType: newchange
Gerrit-Change-Id: I341e46463e654ae0a086cd49af18aa336b86cb6e
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Saggi Mizrahi smizrahi@redhat.com

    

Change in vdsm[master]: jsonrpc: Optimize stomp encoder\decoder