{"action":"created","comment":{"url":"https://api.github.com/repos/pyacq/pyacq/pulls/comments/59816502","id":59816502,"diff_hunk":"@@ -285,84 +284,157 @@ class PlainDataSender:\n     def __init__(self, socket, params):\n         self.socket = socket\n         self.params = params\n-        self.copy = False\n-        #self.copy = not self.params['protocol'] in ('inproc', 'tcp', 'ipc')\n-        \n         self.funcs = []\n-        # compression\n-        if self.params['compression'] == '':\n-            self.funcs.append(self.to_bytes)\n-        elif self.params['compression'] == 'blosc-blosclz':\n-            # cname for ['blosclz', 'lz4', 'lz4hc', 'snappy', 'zlib']\n-            self.funcs.append(self.compress_blosclz)\n-        elif self.params['compression'] == 'blosc-lz4':\n-            self.funcs.append(self.compress_blosclz4)        \n-    \n-    def to_bytes(self, index, data):\n-        if not data.flags['C_CONTIGUOUS']:\n-            # if not C continuous return transform to bytes\n-            # so this is a copy\n-            return index, data.tostring(order='C')\n-        else:\n-            # if C continuous let's keep  data as numpy in case of copy=True\n-            # zmq shoudla avoid copy\n-            return index, data\n     \n-    def compress_blosclz(self, index, data):\n-        assert HAVE_BLOSC, \"Cannot use blosclz compression; blosc package is not importable.\"\n-        data = blosc.pack_array(data, cname='blosclz')\n-        return index, data\n-    \n-    def compress_blosclz4(self, index, data):\n-        assert HAVE_BLOSC, \"Cannot use blosclz4 compression; blosc package is not importable.\"\n-        data = blosc.pack_array(data, cname='lz4')\n-        return index, data\n-    \n-    def send(self, index, data, autoswapaxes = True):\n+    def send(self, index, data):\n+        # optional pre-processing before send\n         if isinstance(data, np.ndarray):\n-            if autoswapaxes and self.params['timeaxis']!=0:\n-                data = data.swapaxes(0, self.params['timeaxis'])\n             for f in self.funcs:\n                 index, data = f(index, data)\n-        self.socket.send_multipart([np.int64(index), data], copy=self.copy)\n+                \n+        # serialize\n+        dtype = data.dtype\n+        shape = data.shape\n+        buf, offset, strides = decompose_array(data)\n+        \n+        # compress\n+        comp = self.params['compression']\n+        if comp.startswith('blosc-'):\n+            assert HAVE_BLOSC, \"Cannot use blosclz4 compression; blosc package is not importable.\"\n+            buf = blosc.compress(buf, data.itemsize, cname=comp[6:])\n+        elif comp != '':\n+            raise ValueError(\"Unknown compression method '%s'\" % comp)\n+        \n+        # Pack and send\n+        stat = struct.pack('!' + 'Q' * (3+len(shape)) + 'q' * len(strides), len(shape), index, offset, *(shape + strides))\n+        copy = self.params.get('copy', False)\n+        self.socket.send_multipart([stat, buf], copy=copy)\n     \n     def close(self):\n         pass\n \n \n-class PlainDataReceiver:\n+class DataReceiver:\n+    def __init__(self, socket, params, ring_size=0, ring_double=True):\n+        self.socket = socket\n+        self.params = params\n+        if ring_size > 0:\n+            ring_shape = (ring_size,) + params['shape'][1:]\n+            self._ring_buffer = RingBuffer(ring_shape, dtype=params['dtype'],\n+                                           double=ring_double, axisorder=params['axisorder'])\n+        else:\n+            self._ring_buffer = None\n+            \n+    def _recv(self):\n+        # must be re-implemented in subclass\n+        raise NotImplementedError()\n+\n+    def recv(self):\n+        index, chunk = self._recv()\n+        if self._ring_buffer is not None:\n+            self._ring_buffer.new_chunk(chunk, index)\n+        return index, chunk\n+\n+    def __getitem__(self, *args):\n+        if self._ring_buffer is None:\n+            raise TypeError(\"DataReceiver indexing requires a ring buffer.\")\n+        return self._ring_buffer[args]\n+    \n+    def get_array_slice(self, index, length):\n+        return self[index:index+length]\n+    \n+    def close(self):\n+        pass\n+    \n+    \n+\n+class PlainDataReceiver(DataReceiver):\n     \"\"\"\n     Helper class to receive data in 2 parts.\n     \n     See PlainDataSender.\n     \"\"\"\n     def __init__(self, socket, params):\n+        ring_size = params['buffer_size']\n+        DataReceiver.__init__(self, socket, params, ring_size=ring_size)\n+    \n+    def _recv(self):\n+        # receive and unpack structure\n+        stat, data = self.socket.recv_multipart()\n+        ndim = struct.unpack('!Q', stat[:8])[0]\n+        stat = struct.unpack('!' + 'Q' * (ndim + 2) + 'q' * ndim, stat[8:])\n+        index = stat[0]\n+        offset = stat[1]\n+        shape = stat[2:2+ndim]\n+        strides = stat[-ndim:]\n+        \n+        # uncompress\n+        comp = self.params['compression']\n+        if comp.startswith('blosc-'):\n+            assert HAVE_BLOSC, \"Cannot use blosc decompression; blosc package is not importable.\"\n+            data = blosc.decompress(data)\n+        elif comp != '':\n+            raise ValueError(\"Unknown compression method '%s'\" % comp)\n+        \n+        # convert to array\n+        dtype = self.params['dtype']\n+        data = np.ndarray(buffer=data, shape=shape,\n+                          strides=strides, offset=offset, dtype=dtype)        \n+        return index, data\n+\n+\n+class SharedMemSender:\n+    \"\"\"\n+    \"\"\"\n+    def __init__(self, socket, params):\n         self.socket = socket\n         self.params = params\n-        \n-        if self.params['compression'] == '':\n-            self.func = self._numpy_fromstring\n-        elif self.params['compression'] in ['blosc-blosclz', 'blosc-lz4']:\n-            self.func = self._uncompress_blosc\n+        self.size = self.params['buffer_size']\n+        dtype = np.dtype(self.params['dtype'])\n+        shape = (self.size,) + self.params['shape'][1:]\n+        self._buffer = RingBuffer(shape=shape, dtype=self.params['dtype'],\n+                                  shmem=True, axisorder=self.params['axisorder'])\n+        self.params['shm_id'] = self._buffer.shm_id\n     \n-    def recv(self, autoswapaxes = True):\n-        m0,m1 = self.socket.recv_multipart()\n-        index = np.fromstring(m0, dtype='int64')[0]\n-        index, data = self.func(index, m1)\n-        if autoswapaxes and self.params['timeaxis']!=0:\n-            data = data.swapaxes(0, self.params['timeaxis'])\n-        return  index, data\n+    def send(self, index, data):\n+        assert data.dtype == self.params['dtype']\n+        shape = data.shape\n+        if self.params['shape'][0] != -1:\n+            assert shape == self.params['shape']\n+        else:\n+            assert shape[1:] == self.params['shape'][1:]\n+ \n+        self._buffer.new_chunk(data, index)\n+        \n+        stat = struct.pack('!' + 'Q' * (2+len(shape)), len(shape), index, *shape)","path":"pyacq/core/stream.py","position":null,"original_position":319,"commit_id":"450122445a0961be56a9cf87362cfd76388bde7a","original_commit_id":"65b23f33f67e7ac3187d01fa4281dda92434c7a4","user":{"login":"campagnola","id":940580,"avatar_url":"https://avatars.githubusercontent.com/u/940580?v=3","gravatar_id":"","url":"https://api.github.com/users/campagnola","html_url":"https://github.com/campagnola","followers_url":"https://api.github.com/users/campagnola/followers","following_url":"https://api.github.com/users/campagnola/following{/other_user}","gists_url":"https://api.github.com/users/campagnola/gists{/gist_id}","starred_url":"https://api.github.com/users/campagnola/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/campagnola/subscriptions","organizations_url":"https://api.github.com/users/campagnola/orgs","repos_url":"https://api.github.com/users/campagnola/repos","events_url":"https://api.github.com/users/campagnola/events{/privacy}","received_events_url":"https://api.github.com/users/campagnola/received_events","type":"User","site_admin":false},"body":"Done.","created_at":"2016-04-15T01:19:19Z","updated_at":"2016-04-15T01:19:19Z","html_url":"https://github.com/pyacq/pyacq/pull/45#discussion_r59816502","pull_request_url":"https://api.github.com/repos/pyacq/pyacq/pulls/45","_links":{"self":{"href":"https://api.github.com/repos/pyacq/pyacq/pulls/comments/59816502"},"html":{"href":"https://github.com/pyacq/pyacq/pull/45#discussion_r59816502"},"pull_request":{"href":"https://api.github.com/repos/pyacq/pyacq/pulls/45"}}},"pull_request":{"url":"https://api.github.com/repos/pyacq/pyacq/pulls/45","id":54068867,"html_url":"https://github.com/pyacq/pyacq/pull/45","diff_url":"https://github.com/pyacq/pyacq/pull/45.diff","patch_url":"https://github.com/pyacq/pyacq/pull/45.patch","issue_url":"https://api.github.com/repos/pyacq/pyacq/issues/45","number":45,"state":"open","locked":false,"title":"Stream performance","user":{"login":"campagnola","id":940580,"avatar_url":"https://avatars.githubusercontent.com/u/940580?v=3","gravatar_id":"","url":"https://api.github.com/users/campagnola","html_url":"https://github.com/campagnola","followers_url":"https://api.github.com/users/campagnola/followers","following_url":"https://api.github.com/users/campagnola/following{/other_user}","gists_url":"https://api.github.com/users/campagnola/gists{/gist_id}","starred_url":"https://api.github.com/users/campagnola/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/campagnola/subscriptions","organizations_url":"https://api.github.com/users/campagnola/orgs","repos_url":"https://api.github.com/users/campagnola/repos","events_url":"https://api.github.com/users/campagnola/events{/privacy}","received_events_url":"https://api.github.com/users/campagnola/received_events","type":"User","site_admin":false},"body":"This doesn't yet change any behavior of pyacq, but implements some benchmarks that should start us on the path to maximizing stream performance.\r\n\r\n* Added `data_transfer_sandbox`, which is a zmq-only demonstration of (I think) the best performance we can expect to see in streams. It minimizes copies and preserves array memory layout (turns out these are mutually dependent goals). This should obviate the need for `autoswapaxes` (if I understand it correctly).\r\n* Moved stream benchmarking to its own file and reworked it a bit such that performance is measured exactly as in the sandbox. Using tcp/plaindata stream gets me ~700MB/s, whereas I can get ~1.3GB/s in the equivalent sandbox test.\r\n\r\n","created_at":"2015-12-18T06:21:07Z","updated_at":"2016-04-15T01:19:19Z","closed_at":null,"merged_at":null,"merge_commit_sha":"e3707496cd51b9c39461cf534e028c295a144c5b","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/pyacq/pyacq/pulls/45/commits","review_comments_url":"https://api.github.com/repos/pyacq/pyacq/pulls/45/comments","review_comment_url":"https://api.github.com/repos/pyacq/pyacq/pulls/comments{/number}","comments_url":"https://api.github.com/repos/pyacq/pyacq/issues/45/comments","statuses_url":"https://api.github.com/repos/pyacq/pyacq/statuses/450122445a0961be56a9cf87362cfd76388bde7a","head":{"label":"campagnola:stream-performance","ref":"stream-performance","sha":"450122445a0961be56a9cf87362cfd76388bde7a","user":{"login":"campagnola","id":940580,"avatar_url":"https://avatars.githubusercontent.com/u/940580?v=3","gravatar_id":"","url":"https://api.github.com/users/campagnola","html_url":"https://github.com/campagnola","followers_url":"https://api.github.com/users/campagnola/followers","following_url":"https://api.github.com/users/campagnola/following{/other_user}","gists_url":"https://api.github.com/users/campagnola/gists{/gist_id}","starred_url":"https://api.github.com/users/campagnola/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/campagnola/subscriptions","organizations_url":"https://api.github.com/users/campagnola/orgs","repos_url":"https://api.github.com/users/campagnola/repos","events_url":"https://api.github.com/users/campagnola/events{/privacy}","received_events_url":"https://api.github.com/users/campagnola/received_events","type":"User","site_admin":false},"repo":{"id":39262280,"name":"pyacq","full_name":"campagnola/pyacq","owner":{"login":"campagnola","id":940580,"avatar_url":"https://avatars.githubusercontent.com/u/940580?v=3","gravatar_id":"","url":"https://api.github.com/users/campagnola","html_url":"https://github.com/campagnola","followers_url":"https://api.github.com/users/campagnola/followers","following_url":"https://api.github.com/users/campagnola/following{/other_user}","gists_url":"https://api.github.com/users/campagnola/gists{/gist_id}","starred_url":"https://api.github.com/users/campagnola/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/campagnola/subscriptions","organizations_url":"https://api.github.com/users/campagnola/orgs","repos_url":"https://api.github.com/users/campagnola/repos","events_url":"https://api.github.com/users/campagnola/events{/privacy}","received_events_url":"https://api.github.com/users/campagnola/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/campagnola/pyacq","description":"Simple Framework for data acquisition (signal, video) in pure python.","fork":true,"url":"https://api.github.com/repos/campagnola/pyacq","forks_url":"https://api.github.com/repos/campagnola/pyacq/forks","keys_url":"https://api.github.com/repos/campagnola/pyacq/keys{/key_id}","collaborators_url":"https://api.github.com/repos/campagnola/pyacq/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/campagnola/pyacq/teams","hooks_url":"https://api.github.com/repos/campagnola/pyacq/hooks","issue_events_url":"https://api.github.com/repos/campagnola/pyacq/issues/events{/number}","events_url":"https://api.github.com/repos/campagnola/pyacq/events","assignees_url":"https://api.github.com/repos/campagnola/pyacq/assignees{/user}","branches_url":"https://api.github.com/repos/campagnola/pyacq/branches{/branch}","tags_url":"https://api.github.com/repos/campagnola/pyacq/tags","blobs_url":"https://api.github.com/repos/campagnola/pyacq/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/campagnola/pyacq/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/campagnola/pyacq/git/refs{/sha}","trees_url":"https://api.github.com/repos/campagnola/pyacq/git/trees{/sha}","statuses_url":"https://api.github.com/repos/campagnola/pyacq/statuses/{sha}","languages_url":"https://api.github.com/repos/campagnola/pyacq/languages","stargazers_url":"https://api.github.com/repos/campagnola/pyacq/stargazers","contributors_url":"https://api.github.com/repos/campagnola/pyacq/contributors","subscribers_url":"https://api.github.com/repos/campagnola/pyacq/subscribers","subscription_url":"https://api.github.com/repos/campagnola/pyacq/subscription","commits_url":"https://api.github.com/repos/campagnola/pyacq/commits{/sha}","git_commits_url":"https://api.github.com/repos/campagnola/pyacq/git/commits{/sha}","comments_url":"https://api.github.com/repos/campagnola/pyacq/comments{/number}","issue_comment_url":"https://api.github.com/repos/campagnola/pyacq/issues/comments{/number}","contents_url":"https://api.github.com/repos/campagnola/pyacq/contents/{+path}","compare_url":"https://api.github.com/repos/campagnola/pyacq/compare/{base}...{head}","merges_url":"https://api.github.com/repos/campagnola/pyacq/merges","archive_url":"https://api.github.com/repos/campagnola/pyacq/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/campagnola/pyacq/downloads","issues_url":"https://api.github.com/repos/campagnola/pyacq/issues{/number}","pulls_url":"https://api.github.com/repos/campagnola/pyacq/pulls{/number}","milestones_url":"https://api.github.com/repos/campagnola/pyacq/milestones{/number}","notifications_url":"https://api.github.com/repos/campagnola/pyacq/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/campagnola/pyacq/labels{/name}","releases_url":"https://api.github.com/repos/campagnola/pyacq/releases{/id}","deployments_url":"https://api.github.com/repos/campagnola/pyacq/deployments","created_at":"2015-07-17T15:53:47Z","updated_at":"2016-03-17T04:29:10Z","pushed_at":"2016-04-15T01:17:14Z","git_url":"git://github.com/campagnola/pyacq.git","ssh_url":"git@github.com:campagnola/pyacq.git","clone_url":"https://github.com/campagnola/pyacq.git","svn_url":"https://github.com/campagnola/pyacq","homepage":null,"size":1096,"stargazers_count":0,"watchers_count":0,"language":"Python","has_issues":false,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":1,"forks":0,"open_issues":1,"watchers":0,"default_branch":"master"}},"base":{"label":"pyacq:master","ref":"master","sha":"8dc90e29efa672251d9fed4fd611d123ce95c9dc","user":{"login":"pyacq","id":12966382,"avatar_url":"https://avatars.githubusercontent.com/u/12966382?v=3","gravatar_id":"","url":"https://api.github.com/users/pyacq","html_url":"https://github.com/pyacq","followers_url":"https://api.github.com/users/pyacq/followers","following_url":"https://api.github.com/users/pyacq/following{/other_user}","gists_url":"https://api.github.com/users/pyacq/gists{/gist_id}","starred_url":"https://api.github.com/users/pyacq/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/pyacq/subscriptions","organizations_url":"https://api.github.com/users/pyacq/orgs","repos_url":"https://api.github.com/users/pyacq/repos","events_url":"https://api.github.com/users/pyacq/events{/privacy}","received_events_url":"https://api.github.com/users/pyacq/received_events","type":"Organization","site_admin":false},"repo":{"id":11476863,"name":"pyacq","full_name":"pyacq/pyacq","owner":{"login":"pyacq","id":12966382,"avatar_url":"https://avatars.githubusercontent.com/u/12966382?v=3","gravatar_id":"","url":"https://api.github.com/users/pyacq","html_url":"https://github.com/pyacq","followers_url":"https://api.github.com/users/pyacq/followers","following_url":"https://api.github.com/users/pyacq/following{/other_user}","gists_url":"https://api.github.com/users/pyacq/gists{/gist_id}","starred_url":"https://api.github.com/users/pyacq/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/pyacq/subscriptions","organizations_url":"https://api.github.com/users/pyacq/orgs","repos_url":"https://api.github.com/users/pyacq/repos","events_url":"https://api.github.com/users/pyacq/events{/privacy}","received_events_url":"https://api.github.com/users/pyacq/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/pyacq/pyacq","description":"Simple Framework for data acquisition (signal, video) in pure python.","fork":false,"url":"https://api.github.com/repos/pyacq/pyacq","forks_url":"https://api.github.com/repos/pyacq/pyacq/forks","keys_url":"https://api.github.com/repos/pyacq/pyacq/keys{/key_id}","collaborators_url":"https://api.github.com/repos/pyacq/pyacq/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/pyacq/pyacq/teams","hooks_url":"https://api.github.com/repos/pyacq/pyacq/hooks","issue_events_url":"https://api.github.com/repos/pyacq/pyacq/issues/events{/number}","events_url":"https://api.github.com/repos/pyacq/pyacq/events","assignees_url":"https://api.github.com/repos/pyacq/pyacq/assignees{/user}","branches_url":"https://api.github.com/repos/pyacq/pyacq/branches{/branch}","tags_url":"https://api.github.com/repos/pyacq/pyacq/tags","blobs_url":"https://api.github.com/repos/pyacq/pyacq/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/pyacq/pyacq/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/pyacq/pyacq/git/refs{/sha}","trees_url":"https://api.github.com/repos/pyacq/pyacq/git/trees{/sha}","statuses_url":"https://api.github.com/repos/pyacq/pyacq/statuses/{sha}","languages_url":"https://api.github.com/repos/pyacq/pyacq/languages","stargazers_url":"https://api.github.com/repos/pyacq/pyacq/stargazers","contributors_url":"https://api.github.com/repos/pyacq/pyacq/contributors","subscribers_url":"https://api.github.com/repos/pyacq/pyacq/subscribers","subscription_url":"https://api.github.com/repos/pyacq/pyacq/subscription","commits_url":"https://api.github.com/repos/pyacq/pyacq/commits{/sha}","git_commits_url":"https://api.github.com/repos/pyacq/pyacq/git/commits{/sha}","comments_url":"https://api.github.com/repos/pyacq/pyacq/comments{/number}","issue_comment_url":"https://api.github.com/repos/pyacq/pyacq/issues/comments{/number}","contents_url":"https://api.github.com/repos/pyacq/pyacq/contents/{+path}","compare_url":"https://api.github.com/repos/pyacq/pyacq/compare/{base}...{head}","merges_url":"https://api.github.com/repos/pyacq/pyacq/merges","archive_url":"https://api.github.com/repos/pyacq/pyacq/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/pyacq/pyacq/downloads","issues_url":"https://api.github.com/repos/pyacq/pyacq/issues{/number}","pulls_url":"https://api.github.com/repos/pyacq/pyacq/pulls{/number}","milestones_url":"https://api.github.com/repos/pyacq/pyacq/milestones{/number}","notifications_url":"https://api.github.com/repos/pyacq/pyacq/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/pyacq/pyacq/labels{/name}","releases_url":"https://api.github.com/repos/pyacq/pyacq/releases{/id}","deployments_url":"https://api.github.com/repos/pyacq/pyacq/deployments","created_at":"2013-07-17T13:38:28Z","updated_at":"2016-04-12T13:41:07Z","pushed_at":"2016-04-15T01:17:14Z","git_url":"git://github.com/pyacq/pyacq.git","ssh_url":"git@github.com:pyacq/pyacq.git","clone_url":"https://github.com/pyacq/pyacq.git","svn_url":"https://github.com/pyacq/pyacq","homepage":null,"size":1076,"stargazers_count":5,"watchers_count":5,"language":"Python","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":8,"mirror_url":null,"open_issues_count":11,"forks":8,"open_issues":11,"watchers":5,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/pyacq/pyacq/pulls/45"},"html":{"href":"https://github.com/pyacq/pyacq/pull/45"},"issue":{"href":"https://api.github.com/repos/pyacq/pyacq/issues/45"},"comments":{"href":"https://api.github.com/repos/pyacq/pyacq/issues/45/comments"},"review_comments":{"href":"https://api.github.com/repos/pyacq/pyacq/pulls/45/comments"},"review_comment":{"href":"https://api.github.com/repos/pyacq/pyacq/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/pyacq/pyacq/pulls/45/commits"},"statuses":{"href":"https://api.github.com/repos/pyacq/pyacq/statuses/450122445a0961be56a9cf87362cfd76388bde7a"}}}}