develop

Author: Andrei-Pozolotin

.gitignore

@@ -102,3 +102,6 @@ venv.bak/
 
 # mypy
 .mypy_cache/
+
+# cffi
+*_cffi.*

.travis.yml

@@ -0,0 +1,21 @@
+dist: xenial
+sudo: true
+language: python
+git:
+  depth: 10
+branches:
+  only:
+  - master
+  - /develop/
+matrix:
+  include:
+  - python: 3.8
+    env: TOXENV=py38
+install:
+- pip install tox devrepo
+script:
+- echo $HOME
+- uname -a
+- ip addr
+- python --version
+- tox

README.md

@@ -0,0 +1,64 @@
+
+## data_pipe
+
+[![Travis Status][travis_icon]][travis_link]
+[![Package Version][pypi_icon]][pypi_link]
+[![Python Versions][python_icon]][python_link]
+
+### Features
+
+Cross Thread Message Pipe:
+* lock-free
+* thread-safe
+* exponential-back-off
+
+Message Transfer End Points:
+* any combination of: sync, trio, curio, asyncio
+
+### Benchmark
+
+Performance on local CPU:
+
+cross-thread
+```
+sync -> sync     @ 1.0 micros/object
+sync -> curio    @ 5.0 micros/object
+sync -> asyncio  @ 6.0 micros/object
+sync -> trio     @ 7.0 micros/object
+```
+
+cross-process
+```
+sync -> sync     @ 0.5 micros/object
+sync -> curio    @ 0.7 micros/object
+sync -> asyncio  @ 0.7 micros/object
+sync -> trio     @ 0.7 micros/object
+```
+
+### Install
+
+To install python package:
+
+```
+sudo pip install data_pipe
+```
+
+### Usage
+
+study:
+* [any_ruptor_perf.py](https://github.com/random-python/data_pipe/blob/master/src/perf/data_pipe_perf/any_ruptor_perf.py)
+
+
+
+
+[travis_icon]: https://travis-ci.org/random-python/data_pipe.svg?branch=master
+[travis_link]: https://travis-ci.org/random-python/data_pipe/builds
+
+[pypi_icon]: https://badge.fury.io/py/data-pipe.svg
+[pypi_link]: https://pypi.python.org/pypi/data-pipe
+
+[python_icon]: https://img.shields.io/pypi/pyversions/data_pipe.svg
+[python_link]: https://pypi.python.org/pypi/data-pipe
+
+[tokei_icon]: https://tokei.rs/b1/github/random-python/data_pipe
+[tokei_link]: https://github.com/random-python/data_pipe/tree/master/src

LICENSE licence

@@ -0,0 +1,8 @@
+
+cffi >= 1.13.2
+
+numba >= 0.46.0
+
+numpy >= 1.17.4
+
+trio >= 0.13.0

readme.md

@@ -0,0 +1,4 @@
+
+trio >= 0.13.0
+
+curio >= 0.9

requirements-dev.txt

@@ -0,0 +1,65 @@
+[metadata]
+
+name = data_pipe
+version = 0.1.5
+
+requires-python = >=3.8
+
+author = Andrei Pozolotin
+author-email = [email protected]
+
+summary = Cross Thread Message Pipe
+
+description-file = readme.md
+description-content-type = text/markdown; charset=UTF-8
+
+home-page = https://github.com/random-python/data_pipe
+
+license = Apache-2
+
+classifier =
+    Development Status :: 4 - Beta
+    Intended Audience :: Developers
+    License :: OSI Approved :: Apache Software License
+    Operating System :: POSIX :: Linux
+    Programming Language :: Python
+    Programming Language :: Python :: 3.8
+    Topic :: Utilities
+    
+
+keywords =
+    cross
+    thread
+    message
+    pipe
+    memory
+    channel
+    trio
+    asyncio
+    threading
+
+[files]
+# none
+
+[options]    
+
+packages =
+    data_pipe
+    data_pipe_test
+    
+package_dir = 
+    data_pipe      = src/main/data_pipe
+    data_pipe_test = src/test/data_pipe_test
+
+[entry_points]
+# none
+
+console_scripts =
+# none
+    
+[pbr]
+
+warnerrors = True
+skip_authors = True
+skip_git_sdist = False
+skip_changelog = True

requirements.txt

@@ -0,0 +1,23 @@
+#!/usr/bin/env python
+
+from setuptools import setup, Extension
+
+setup(
+
+    pbr=True,
+
+    setup_requires=[
+        "pbr",
+        "cython"
+    ],
+
+    ext_modules=[
+        Extension(
+            name="data_pipe.runtime_library",
+            sources=[
+                "src/main/data_pipe/runtime_library.pyx",
+            ],
+        ),
+    ]
+
+)

setup.cfg

@@ -0,0 +1,6 @@
+
+# generated resources
+
+*.c
+*.so
+*.html

setup.py

@@ -0,0 +1,115 @@
+"""
+"""
+
+import time
+import math
+import trio
+import curio
+import asyncio
+
+from typing import Any
+from dataclasses import dataclass, field
+from trio._core._run import GLOBAL_RUN_CONTEXT
+
+from data_pipe.any_index import AnyIndexBase
+
+
+class AnyBufferBase:
+    ""
+
+    default_wait_size = 15
+    default_wait_list = [10e-6 * (2 ** index) for index in range(default_wait_size - 1, -1, -1)]
+
+    @classmethod
+    def has_trio_loop(cls) -> bool:
+        ""
+        return hasattr(GLOBAL_RUN_CONTEXT, "runner")
+
+    @classmethod
+    def has_curio_loop(cls) -> bool:
+        ""
+        return curio.meta.curio_running()
+
+    @classmethod
+    def has_asyncio_loop(cls) -> bool:
+        ""
+        return asyncio._get_running_loop() is not None
+
+    @classmethod
+    async def perform_wait(cls, wait_time:float) -> None:
+        ""
+        if cls.has_trio_loop():
+            await trio.sleep(wait_time)
+        elif cls.has_curio_loop():
+            await curio.sleep(wait_time)
+        elif cls.has_asyncio_loop():
+            await asyncio.sleep(wait_time)
+        else:
+            assert False, f"wrong async"
+
+    @classmethod
+    def perform_wait_sync(cls, wait_time:float) -> None:
+        ""
+        time.sleep(wait_time)
+
+
+@dataclass(frozen=True)
+class AnyBufferCore(AnyBufferBase):
+    ""
+
+    ring_size:int = field(default=128)
+    wait_list:list = field(default_factory=lambda:AnyBufferBase.default_wait_list, repr=False)
+
+    _ring_array:list = field(default=None, repr=False, init=False)
+    _ring_index:AnyIndexBase = field(default_factory=AnyIndexBase, repr=False, init=False)
+
+    def __post_init__(self):
+        ""
+        assert math.log(self.ring_size, 2).is_integer(), f"wrong ring_size={self.ring_size}"
+        _ring_array = [ None for _ in range(self.ring_size) ]
+        object.__setattr__(self, '_ring_array', _ring_array)
+        self._ring_index.setup(ring_size=self.ring_size, wait_size=len(self.wait_list))
+
+    async def invoke_reader(self) -> Any:
+        ""
+        while True:
+            index = self._ring_index.next_reader()
+            if index < 0:
+                await self.perform_wait(self.wait_list[index])
+            else:
+                return self._ring_array[index]
+
+    def invoke_reader_sync(self) -> Any:
+        ""
+        while True:
+            index = self._ring_index.next_reader()
+            if index < 0:
+                self.perform_wait_sync(self.wait_list[index])
+            else:
+                return self._ring_array[index]
+
+    async def invoke_writer(self, value:Any) -> None:
+        ""
+        while True:
+            index = self._ring_index.next_writer()
+            if index < 0:
+                await self.perform_wait(self.wait_list[index])
+            else:
+                self._ring_array[index] = value
+                return
+
+    def invoke_writer_sync(self, value:Any) -> None:
+        ""
+        while True:
+            index = self._ring_index.next_writer()
+            if index < 0:
+                self.perform_wait_sync(self.wait_list[index])
+            else:
+                self._ring_array[index] = value
+                return
+    #
+    #
+    #
+
+    get = invoke_reader_sync
+    put = invoke_writer_sync

src/main/data_pipe/.gitignore

@@ -0,0 +1,15 @@
+"""
+"""
+
+
+class AnyIndexBase:
+    ""
+
+    def setup(self, *, ring_size:int, wait_size:int) -> None:
+        ""
+
+    def next_reader(self) -> int:
+        ""
+
+    def next_writer(self) -> int:
+        ""

src/main/data_pipe/__init__.py

@@ -0,0 +1,14 @@
+"""
+"""
+
+from dataclasses import dataclass, field
+
+from data_pipe.any_buffer import AnyBufferCore
+from data_pipe.basic_index import BasicIndex
+
+
+@dataclass(frozen=True)
+class BasicBuffer(AnyBufferCore):
+    ""
+
+    _ring_index:BasicIndex = field(default_factory=BasicIndex, repr=False)