django/asgi_ipc

IPC-based ASGI channel layer

Django

Django is a high-level Python web framework that encourages rapid development and clean, pragmatic design. Thanks for checking it out.

All documentation is in the "docs" directory and online at https://docs.djangoproject.com/en/stable/. If you're just getting started, here's how we recommend you read the docs:

• First, read docs/intro/install.txt for instructions on installing Django.
• Next, work through the tutorials in order (docs/intro/tutorial01.txt, docs/intro/tutorial02.txt, etc.).
• If you want to set up an actual deployment server, read docs/howto/deployment/index.txt for instructions.
• You'll probably want to read through the topical guides (in docs/topics) next; from there you can jump to the HOWTOs (in docs/howto) for specific problems, and check out the reference (docs/ref) for gory details.
• See docs/README for instructions on building an HTML version of the docs.

Docs are updated rigorously. If you find any problems in the docs, or think they should be clarified in any way, please take 30 seconds to fill out a ticket here: https://code.djangoproject.com/newticket

To get more help:

• Join the #django channel on irc.libera.chat. Lots of helpful people hang out there. Webchat is available.
• Join the django-users mailing list, or read the archives, at https://groups.google.com/group/django-users.
• Join the Django Discord community.
• Join the community on the Django Forum.

To contribute to Django:

• Check out https://docs.djangoproject.com/en/dev/internals/contributing/ for information about getting involved.

To run Django's test suite:

• Follow the instructions in the "Unit tests" section of docs/internals/contributing/writing-code/unit-tests.txt, published online at https://docs.djangoproject.com/en/dev/internals/contributing/writing-code/unit-tests/#running-the-unit-tests

Supporting the Development of Django

Django's development depends on your contributions.

If you depend on Django, remember to support the Django Software Foundation: https://www.djangoproject.com/fundraising/

asgi_ipc's Issues

Starting daphne causes ValueError: The size is invalid or the memory is read-only

I get this every single time I try to start a server with daphne cjworkbench.asgi:channel_layer:

Traceback (most recent call last):
  File "/Users/jonathanstray/anaconda/lib/python3.5/site-packages/asgi_ipc.py", line 208, in __init__
    size=self.size,
ValueError: The size is invalid or the memory is read-only

And then several other exceptions while handling that one. This is on Mac OS. Maybe is mmap not supported on this OS?

My settings:

CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "asgi_ipc.IPCChannelLayer",
        "ROUTING": "cjworkbench.routing.channel_routing",
        "CONFIG": {
            "prefix": "cjworkbench",
        },
    },
}

Address slow performance

The load tests (https://github.com/andrewgodwin/channels/blob/master/loadtesting/2016-09-06/README.rst) showed very poor performance for the IPC backend. We should investigate why and see if we can get it a bit faster - my suspicion is that the global semaphore lock strategy is quite slow.

Asgiref 2.0.1 version conflict

Hi,

After upgrading stuff via pip, asgiref-2.0.1 was installed
asgi_ipc is the latest version as well : 1.4.2
Loading my project I got the error below
Obviously setup.py contains as dependency : asgiref~=1.1.2
ERROR [celery.utils.dispatch.signal:276] Signal handler <bound method DjangoFixup.on_import_modules of <celery.fixups.django.DjangoFixup object at 0x757c89b0>> raised: ContextualVersionConflict(asgiref 2.0.1 (/usr/lib/python3.6/site-packages), Requirement.parse('asgiref~=1.1.2'), {'asgi-ipc'}) Traceback (most recent call last): File "/usr/lib/python3.6/site-packages/celery/utils/dispatch/signal.py", line 271, in send response = receiver(signal=self, sender=sender, **named) File "/usr/lib/python3.6/site-packages/celery/fixups/django.py", line 84, in on_import_modules self.worker_fixup.validate_models() File "/usr/lib/python3.6/site-packages/celery/fixups/django.py", line 122, in validate_models self.django_setup() File "/usr/lib/python3.6/site-packages/celery/fixups/django.py", line 118, in django_setup django.setup() File "/usr/lib/python3.6/site-packages/django/__init__.py", line 27, in setup apps.populate(settings.INSTALLED_APPS) File "/usr/lib/python3.6/site-packages/django/apps/registry.py", line 116, in populate app_config.ready() File "/usr/lib/python3.6/site-packages/channels/apps.py", line 14, in ready check_all() File "/usr/lib/python3.6/site-packages/channels/package_checks.py", line 22, in check_all module = importlib.import_module(package) File "/usr/lib/python3.6/importlib/__init__.py", line 126, in import_module return _bootstrap._gcd_import(name[level:], package, level) File "<frozen importlib._bootstrap>", line 994, in _gcd_import File "<frozen importlib._bootstrap>", line 971, in _find_and_load File "<frozen importlib._bootstrap>", line 955, in _find_and_load_unlocked File "<frozen importlib._bootstrap>", line 665, in _load_unlocked File "<frozen importlib._bootstrap_external>", line 678, in exec_module File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed File "/usr/lib/python3.6/site-packages/asgi_ipc/__init__.py", line 2, in <module> __version__ = pkg_resources.require('asgi_ipc')[0].version File "/usr/lib/python3.6/site-packages/pkg_resources/__init__.py", line 984, in require needed = self.resolve(parse_requirements(requirements)) File "/usr/lib/python3.6/site-packages/pkg_resources/__init__.py", line 875, in resolve raise VersionConflict(dist, req).with_context(dependent_req) pkg_resources.ContextualVersionConflict: (asgiref 2.0.1 (/usr/lib/python3.6/site-packages), Requirement.parse('asgiref~=1.1.2'), {'asgi-ipc'})

./manage.py runworker randomly crashes at startup

Hello!

I'm trying to use asgi_ipc as channel backends but if sometimes it works :

$ ./manage.py runworker
2016-09-29 12:31:35,258 - INFO - runworker - Running worker against channel layer default (asgi_ipc.IPCChannelLayer)
2016-09-29 12:31:35,259 - INFO - worker - Listening on channels http.request, websocket.connect, websocket.disconnect, websocket.receive

But sometimes (3 times out of 4) it crashes :

$ ./manage.py runworker
Traceback (most recent call last):
  File "/path/to/site-packages/asgi_ipc.py", line 208, in __init__
    size=self.size,
ValueError: The size is invalid or the memory is read-only

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "./manage.py", line 10, in <module>
    execute_from_command_line(sys.argv)
  File "/path/to/site-packages/django/core/management/__init__.py", line 353, in execute_from_command_line
    utility.execute()
  File "/path/to/site-packages/django/core/management/__init__.py", line 345, in execute
    self.fetch_command(subcommand).run_from_argv(self.argv)
  File "/path/to/site-packages/django/core/management/base.py", line 348, in run_from_argv
    self.execute(*args, **cmd_options)
  File "/path/to/site-packages/django/core/management/base.py", line 399, in execute
    output = self.handle(*args, **options)
  File "/path/to/site-packages/channels/management/commands/runworker.py", line 36, in handle
    self.channel_layer = channel_layers[options.get("layer", DEFAULT_CHANNEL_LAYER)]
  File "/path/to/site-packages/channels/asgi.py", line 53, in __getitem__
    self.backends[key] = self.make_backend(key)
  File "/path/to/site-packages/channels/asgi.py", line 44, in make_backend
    asgi_layer = backend_class(**self.configs[name].get("CONFIG", {}))
  File "/path/to/site-packages/asgi_ipc.py", line 43, in __init__
    self.channel_store = MemoryDict("/%s-chan" % self.prefix, size=channel_memory)
  File "/path/to/site-packages/asgi_ipc.py", line 213, in __init__
    "Error was: %s" % e
ValueError: Unable to allocate shared memory segment (potentially out of memory).
Error was: The size is invalid or the memory is read-only
Exception ignored in: <bound method MemoryDatastructure.__del__ of <asgi_ipc.MemoryDict object at 0x104d48438>>
Traceback (most recent call last):
  File "/path/to/site-packages/asgi_ipc.py", line 294, in __del__
    self.mmap.close()
AttributeError: 'MemoryDict' object has no attribute 'mmap'

I used the following CHANNEL_LAYERS value :

CHANNEL_LAYERS = {
    "default" : {
        "BACKEND" : "asgi_ipc.IPCChannelLayer",
        "CONFIG" : {
            "prefix" : "ws",
        },
        "ROUTING": "ws.routing.channel_routing",
    },
}

And installed asgi_ipc from sources, the last commit being :

f22c60a

I've seen the issue #4 but I don't know it it is related. Anyway the PR used to solve this one doesn't prevent this problem for me.

I have no problem if I use an other channel backend (tested with asgi_redis).

Cannot receive on channel after restarting. Bug?

I have an issue and I am not sure if that is by design. When my program restarts it does not receive any more messages on a channel.

I prepared an example which you find further down.. At the 30th iteration I am simulating the restart of the program by simply doing this:

channel_layer_receive = None
channel_layer_receive = asgi.IPCChannelLayer(prefix="my_prefix")

After that the script keeps printing (None, None) although it is still sending on the other channel layer.

Is that by design or a bug?

Example

import asgi_ipc as asgi

channel_layer_receive = asgi.IPCChannelLayer(prefix="my_prefix")
channel_layer_send = asgi.IPCChannelLayer(prefix="my_prefix")

i = 0

while i < 10:
    i += 1
    msg = "Message %s" % i
    try:
        channel_layer_send.send("my_channel", {"text": msg})
        print("Sending %s" % msg)
    except asgi.BaseChannelLayer.ChannelFull:
        print("Dropped %s" % msg)
        pass

    print(channel_layer_receive.receive(["my_channel"]))

    if i == 5:
        channel_layer_receive = None
        channel_layer_receive = asgi.IPCChannelLayer(prefix="my_prefix")

print("Done!")

Output

Sending Message 1
('my_channel', {'text': 'Message 1'})
Sending Message 2
('my_channel', {'text': 'Message 2'})
Sending Message 3
('my_channel', {'text': 'Message 3'})
Sending Message 4
('my_channel', {'text': 'Message 4'})
Sending Message 5
('my_channel', {'text': 'Message 5'})
Sending Message 6
(None, None)
Sending Message 7
(None, None)
Sending Message 8
(None, None)
Sending Message 9
(None, None)
Sending Message 10
(None, None)
Done!
Exception ignored in: <bound method MemoryDict.__del__ of <asgi_ipc.MemoryDict object at 0x7f59a875c390>>
Traceback (most recent call last):
  File "/usr/local/lib/python3.4/dist-packages/asgi_ipc.py", line 311, in __del__
posix_ipc.ExistentialError: No shared memory exists with the specified name
Exception ignored in: <bound method MemoryDict.__del__ of <asgi_ipc.MemoryDict object at 0x7f59a88a6748>>
Traceback (most recent call last):
  File "/usr/local/lib/python3.4/dist-packages/asgi_ipc.py", line 311, in __del__
posix_ipc.ExistentialError: No shared memory exists with the specified name

Potential message duplication

Apparently in some conditions messages can be duplicated during send. Originally reported here django/channels#398

Unable to use layer inside docker container.

This is a really weird situation.

I hit this thing while trying to fix asgi_rabbitmq build. This is what I've got when I try to use IPC layer inside Docker container.

$ docker-compose run --rm py36dj111 /code/.tox3.6.1/py36-django111/bin/python -i
Starting asgirabbitmq_rabbitmq_1 ... done
Python 3.6.1 (default, Mar 23 2017, 02:34:11)
[GCC 4.9.2] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> from asgi_ipc import IPCChannelLayer
>>> layer = IPCChannelLayer()
>>> layer.send('foo', {'baz': 'bar'})
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/code/.tox3.6.1/py36-django111/lib/python3.6/site-packages/asgi_ipc/core.py", line 61, in send
    channel_size = self.message_store.length(channel)
  File "/code/.tox3.6.1/py36-django111/lib/python3.6/site-packages/asgi_ipc/store.py", line 134, in length
    with self.mutate_value() as value:
  File "/usr/local/lib/python3.6/contextlib.py", line 82, in __enter__
    return next(self.gen)
  File "/code/.tox3.6.1/py36-django111/lib/python3.6/site-packages/asgi_ipc/store.py", line 50, in mutate_value
    value = pickle.load(self.mmap)
_pickle.UnpicklingError: invalid load key, '\x00'.
>>> layer = IPCChannelLayer()
>>> layer.flush()
>>> layer.send('foo', {'baz': 'bar'})

As you can see, send will be successful only after first use of flush. I can't reproduce this behavior outside of Docker container. It works fine on my machine. But Travis builds use Docker for the build, so exactly same error happens in the CI.

https://travis-ci.org/proofit404/asgi_rabbitmq/jobs/236879296#L4355-L4356

It is curious because asgi_ipc and asgi_redis builds passed without any trouble.

Any suggestions for future research?

TypeError: Struct() argument 1 must be string, not unicode while testing

Using CentOS 7 (ships with Python v2.7.5) and get the error: TypeError: Struct() argument 1 must be string, not unicode (see below).

================================= test session starts ==================================
platform linux2 -- Python 2.7.5, pytest-3.0.6, py-1.4.31, pluggy-0.4.0
rootdir: /builddir/build/BUILD/asgi_ipc-1.2.0, inifile: 
plugins: hypothesis-3.6.1
collected 29 items 

test_asgi_ipc.py FFFFFFFFFFFFFFFssssssssssssss

======================================= FAILURES =======================================
_____________________________ IPCLayerTests.test_capacity ______________________________

self = <test_asgi_ipc.IPCLayerTests testMethod=test_capacity>

    def setUp(self):
        if self.channel_layer is None:
            raise ValueError("You must define 'channel_layer' when subclassing the conformance tests.")
        if self.expiry_delay is None:
            raise ValueError("You must define 'expiry_delay' when subclassing the conformance tests.")
        if "flush" in self.channel_layer.extensions:
>           self.channel_layer.flush()

/usr/lib/python2.7/site-packages/asgiref/conformance.py:43: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
asgi_ipc.py:164: in flush
    self.channel_store.flush()
asgi_ipc.py:299: in flush
    self._set_value(self.datatype())
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <asgi_ipc.MemoryDict object at 0x241ab10>, value = {}

    def _set_value(self, value):
        assert isinstance(value, self.datatype)
        try:
            self.semaphore.acquire(self.death_timeout)
        except posix_ipc.BusyError:
            self._reset()
            self.semaphore.acquire(0)
        try:
            self.mmap.seek(0)
            self.mmap.write(self.signature)
            towrite = msgpack.packb(value, use_bin_type=True)
>           self.mmap.write(struct.pack("!I", len(towrite)))
E           TypeError: Struct() argument 1 must be string, not unicode

asgi_ipc.py:265: TypeError
____________________________ IPCLayerTests.test_exceptions _____________________________

self = <test_asgi_ipc.IPCLayerTests testMethod=test_exceptions>

    def setUp(self):
        if self.channel_layer is None:
            raise ValueError("You must define 'channel_layer' when subclassing the conformance tests.")
        if self.expiry_delay is None:
            raise ValueError("You must define 'expiry_delay' when subclassing the conformance tests.")
        if "flush" in self.channel_layer.extensions:
>           self.channel_layer.flush()

/usr/lib/python2.7/site-packages/asgiref/conformance.py:43: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
asgi_ipc.py:164: in flush
    self.channel_store.flush()
asgi_ipc.py:299: in flush
    self._set_value(self.datatype())
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <asgi_ipc.MemoryDict object at 0x241ab10>, value = {}

    def _set_value(self, value):
        assert isinstance(value, self.datatype)
        try:
            self.semaphore.acquire(self.death_timeout)
        except posix_ipc.BusyError:
            self._reset()
            self.semaphore.acquire(0)
        try:
            self.mmap.seek(0)
            self.mmap.write(self.signature)
            towrite = msgpack.packb(value, use_bin_type=True)
>           self.mmap.write(struct.pack("!I", len(towrite)))
E           TypeError: Struct() argument 1 must be string, not unicode

asgi_ipc.py:265: TypeError
_______________________________ IPCLayerTests.test_flush _______________________________

self = <test_asgi_ipc.IPCLayerTests testMethod=test_flush>

    def setUp(self):
        if self.channel_layer is None:
            raise ValueError("You must define 'channel_layer' when subclassing the conformance tests.")
        if self.expiry_delay is None:
            raise ValueError("You must define 'expiry_delay' when subclassing the conformance tests.")
        if "flush" in self.channel_layer.extensions:
>           self.channel_layer.flush()

/usr/lib/python2.7/site-packages/asgiref/conformance.py:43: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
asgi_ipc.py:164: in flush
    self.channel_store.flush()
asgi_ipc.py:299: in flush
    self._set_value(self.datatype())
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <asgi_ipc.MemoryDict object at 0x241ab10>, value = {}

    def _set_value(self, value):
        assert isinstance(value, self.datatype)
        try:
            self.semaphore.acquire(self.death_timeout)
        except posix_ipc.BusyError:
            self._reset()
            self.semaphore.acquire(0)
        try:
            self.mmap.seek(0)
            self.mmap.write(self.signature)
            towrite = msgpack.packb(value, use_bin_type=True)
>           self.mmap.write(struct.pack("!I", len(towrite)))
E           TypeError: Struct() argument 1 must be string, not unicode

asgi_ipc.py:265: TypeError
___________________________ IPCLayerTests.test_flush_groups ____________________________

self = <test_asgi_ipc.IPCLayerTests testMethod=test_flush_groups>

    def setUp(self):
        if self.channel_layer is None:
            raise ValueError("You must define 'channel_layer' when subclassing the conformance tests.")
        if self.expiry_delay is None:
            raise ValueError("You must define 'expiry_delay' when subclassing the conformance tests.")
        if "flush" in self.channel_layer.extensions:
>           self.channel_layer.flush()

/usr/lib/python2.7/site-packages/asgiref/conformance.py:43: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
asgi_ipc.py:164: in flush
    self.channel_store.flush()
asgi_ipc.py:299: in flush
    self._set_value(self.datatype())
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <asgi_ipc.MemoryDict object at 0x241ab10>, value = {}

    def _set_value(self, value):
        assert isinstance(value, self.datatype)
        try:
            self.semaphore.acquire(self.death_timeout)
        except posix_ipc.BusyError:
            self._reset()
            self.semaphore.acquire(0)
        try:
            self.mmap.seek(0)
            self.mmap.write(self.signature)
            towrite = msgpack.packb(value, use_bin_type=True)
>           self.mmap.write(struct.pack("!I", len(towrite)))
E           TypeError: Struct() argument 1 must be string, not unicode

asgi_ipc.py:265: TypeError
__________________________ IPCLayerTests.test_group_channels ___________________________

self = <test_asgi_ipc.IPCLayerTests testMethod=test_group_channels>

    def setUp(self):
        if self.channel_layer is None:
            raise ValueError("You must define 'channel_layer' when subclassing the conformance tests.")
        if self.expiry_delay is None:
            raise ValueError("You must define 'expiry_delay' when subclassing the conformance tests.")
        if "flush" in self.channel_layer.extensions:
>           self.channel_layer.flush()

/usr/lib/python2.7/site-packages/asgiref/conformance.py:43: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
asgi_ipc.py:164: in flush
    self.channel_store.flush()
asgi_ipc.py:299: in flush
    self._set_value(self.datatype())
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <asgi_ipc.MemoryDict object at 0x241ab10>, value = {}

    def _set_value(self, value):
        assert isinstance(value, self.datatype)
        try:
            self.semaphore.acquire(self.death_timeout)
        except posix_ipc.BusyError:
            self._reset()
            self.semaphore.acquire(0)
        try:
            self.mmap.seek(0)
            self.mmap.write(self.signature)
            towrite = msgpack.packb(value, use_bin_type=True)
>           self.mmap.write(struct.pack("!I", len(towrite)))
E           TypeError: Struct() argument 1 must be string, not unicode

asgi_ipc.py:265: TypeError
___________________________ IPCLayerTests.test_group_expiry ____________________________

self = <test_asgi_ipc.IPCLayerTests testMethod=test_group_expiry>

    def setUp(self):
        if self.channel_layer is None:
            raise ValueError("You must define 'channel_layer' when subclassing the conformance tests.")
        if self.expiry_delay is None:
            raise ValueError("You must define 'expiry_delay' when subclassing the conformance tests.")
        if "flush" in self.channel_layer.extensions:
>           self.channel_layer.flush()

/usr/lib/python2.7/site-packages/asgiref/conformance.py:43: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
asgi_ipc.py:164: in flush
    self.channel_store.flush()
asgi_ipc.py:299: in flush
    self._set_value(self.datatype())
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <asgi_ipc.MemoryDict object at 0x241ab10>, value = {}

    def _set_value(self, value):
        assert isinstance(value, self.datatype)
        try:
            self.semaphore.acquire(self.death_timeout)
        except posix_ipc.BusyError:
            self._reset()
            self.semaphore.acquire(0)
        try:
            self.mmap.seek(0)
            self.mmap.write(self.signature)
            towrite = msgpack.packb(value, use_bin_type=True)
>           self.mmap.write(struct.pack("!I", len(towrite)))
E           TypeError: Struct() argument 1 must be string, not unicode

asgi_ipc.py:265: TypeError
______________________________ IPCLayerTests.test_groups _______________________________

self = <test_asgi_ipc.IPCLayerTests testMethod=test_groups>

    def setUp(self):
        if self.channel_layer is None:
            raise ValueError("You must define 'channel_layer' when subclassing the conformance tests.")
        if self.expiry_delay is None:
            raise ValueError("You must define 'expiry_delay' when subclassing the conformance tests.")
        if "flush" in self.channel_layer.extensions:
>           self.channel_layer.flush()

/usr/lib/python2.7/site-packages/asgiref/conformance.py:43: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
asgi_ipc.py:164: in flush
    self.channel_store.flush()
asgi_ipc.py:299: in flush
    self._set_value(self.datatype())
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <asgi_ipc.MemoryDict object at 0x241ab10>, value = {}

    def _set_value(self, value):
        assert isinstance(value, self.datatype)
        try:
            self.semaphore.acquire(self.death_timeout)
        except posix_ipc.BusyError:
            self._reset()
            self.semaphore.acquire(0)
        try:
            self.mmap.seek(0)
            self.mmap.write(self.signature)
            towrite = msgpack.packb(value, use_bin_type=True)
>           self.mmap.write(struct.pack("!I", len(towrite)))
E           TypeError: Struct() argument 1 must be string, not unicode

asgi_ipc.py:265: TypeError
___________________ IPCLayerTests.test_message_alteration_after_send ___________________

self = <test_asgi_ipc.IPCLayerTests testMethod=test_message_alteration_after_send>

    def setUp(self):
        if self.channel_layer is None:
            raise ValueError("You must define 'channel_layer' when subclassing the conformance tests.")
        if self.expiry_delay is None:
            raise ValueError("You must define 'expiry_delay' when subclassing the conformance tests.")
        if "flush" in self.channel_layer.extensions:
>           self.channel_layer.flush()

/usr/lib/python2.7/site-packages/asgiref/conformance.py:43: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
asgi_ipc.py:164: in flush
    self.channel_store.flush()
asgi_ipc.py:299: in flush
    self._set_value(self.datatype())
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <asgi_ipc.MemoryDict object at 0x241ab10>, value = {}

    def _set_value(self, value):
        assert isinstance(value, self.datatype)
        try:
            self.semaphore.acquire(self.death_timeout)
        except posix_ipc.BusyError:
            self._reset()
            self.semaphore.acquire(0)
        try:
            self.mmap.seek(0)
            self.mmap.write(self.signature)
            towrite = msgpack.packb(value, use_bin_type=True)
>           self.mmap.write(struct.pack("!I", len(towrite)))
E           TypeError: Struct() argument 1 must be string, not unicode

asgi_ipc.py:265: TypeError
__________________________ IPCLayerTests.test_message_expiry ___________________________

self = <test_asgi_ipc.IPCLayerTests testMethod=test_message_expiry>

    def setUp(self):
        if self.channel_layer is None:
            raise ValueError("You must define 'channel_layer' when subclassing the conformance tests.")
        if self.expiry_delay is None:
            raise ValueError("You must define 'expiry_delay' when subclassing the conformance tests.")
        if "flush" in self.channel_layer.extensions:
>           self.channel_layer.flush()

/usr/lib/python2.7/site-packages/asgiref/conformance.py:43: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
asgi_ipc.py:164: in flush
    self.channel_store.flush()
asgi_ipc.py:299: in flush
    self._set_value(self.datatype())
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <asgi_ipc.MemoryDict object at 0x241ab10>, value = {}

    def _set_value(self, value):
        assert isinstance(value, self.datatype)
        try:
            self.semaphore.acquire(self.death_timeout)
        except posix_ipc.BusyError:
            self._reset()
            self.semaphore.acquire(0)
        try:
            self.mmap.seek(0)
            self.mmap.write(self.signature)
            towrite = msgpack.packb(value, use_bin_type=True)
>           self.mmap.write(struct.pack("!I", len(towrite)))
E           TypeError: Struct() argument 1 must be string, not unicode

asgi_ipc.py:265: TypeError
_______________________ IPCLayerTests.test_new_channel_failures ________________________

self = <test_asgi_ipc.IPCLayerTests testMethod=test_new_channel_failures>

    def setUp(self):
        if self.channel_layer is None:
            raise ValueError("You must define 'channel_layer' when subclassing the conformance tests.")
        if self.expiry_delay is None:
            raise ValueError("You must define 'expiry_delay' when subclassing the conformance tests.")
        if "flush" in self.channel_layer.extensions:
>           self.channel_layer.flush()

/usr/lib/python2.7/site-packages/asgiref/conformance.py:43: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
asgi_ipc.py:164: in flush
    self.channel_store.flush()
asgi_ipc.py:299: in flush
    self._set_value(self.datatype())
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <asgi_ipc.MemoryDict object at 0x241ab10>, value = {}

    def _set_value(self, value):
        assert isinstance(value, self.datatype)
        try:
            self.semaphore.acquire(self.death_timeout)
        except posix_ipc.BusyError:
            self._reset()
            self.semaphore.acquire(0)
        try:
            self.mmap.seek(0)
            self.mmap.write(self.signature)
            towrite = msgpack.packb(value, use_bin_type=True)
>           self.mmap.write(struct.pack("!I", len(towrite)))
E           TypeError: Struct() argument 1 must be string, not unicode

asgi_ipc.py:265: TypeError
____________________ IPCLayerTests.test_new_channel_single_process _____________________

self = <test_asgi_ipc.IPCLayerTests testMethod=test_new_channel_single_process>

    def setUp(self):
        if self.channel_layer is None:
            raise ValueError("You must define 'channel_layer' when subclassing the conformance tests.")
        if self.expiry_delay is None:
            raise ValueError("You must define 'expiry_delay' when subclassing the conformance tests.")
        if "flush" in self.channel_layer.extensions:
>           self.channel_layer.flush()

/usr/lib/python2.7/site-packages/asgiref/conformance.py:43: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
asgi_ipc.py:164: in flush
    self.channel_store.flush()
asgi_ipc.py:299: in flush
    self._set_value(self.datatype())
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <asgi_ipc.MemoryDict object at 0x241ab10>, value = {}

    def _set_value(self, value):
        assert isinstance(value, self.datatype)
        try:
            self.semaphore.acquire(self.death_timeout)
        except posix_ipc.BusyError:
            self._reset()
            self.semaphore.acquire(0)
        try:
            self.mmap.seek(0)
            self.mmap.write(self.signature)
            towrite = msgpack.packb(value, use_bin_type=True)
>           self.mmap.write(struct.pack("!I", len(towrite)))
E           TypeError: Struct() argument 1 must be string, not unicode

asgi_ipc.py:265: TypeError
_____________________ IPCLayerTests.test_new_channel_single_reader _____________________

self = <test_asgi_ipc.IPCLayerTests testMethod=test_new_channel_single_reader>

    def setUp(self):
        if self.channel_layer is None:
            raise ValueError("You must define 'channel_layer' when subclassing the conformance tests.")
        if self.expiry_delay is None:
            raise ValueError("You must define 'expiry_delay' when subclassing the conformance tests.")
        if "flush" in self.channel_layer.extensions:
>           self.channel_layer.flush()

/usr/lib/python2.7/site-packages/asgiref/conformance.py:43: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
asgi_ipc.py:164: in flush
    self.channel_store.flush()
asgi_ipc.py:299: in flush
    self._set_value(self.datatype())
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <asgi_ipc.MemoryDict object at 0x241ab10>, value = {}

    def _set_value(self, value):
        assert isinstance(value, self.datatype)
        try:
            self.semaphore.acquire(self.death_timeout)
        except posix_ipc.BusyError:
            self._reset()
            self.semaphore.acquire(0)
        try:
            self.mmap.seek(0)
            self.mmap.write(self.signature)
            towrite = msgpack.packb(value, use_bin_type=True)
>           self.mmap.write(struct.pack("!I", len(towrite)))
E           TypeError: Struct() argument 1 must be string, not unicode

asgi_ipc.py:265: TypeError
_____________________________ IPCLayerTests.test_send_recv _____________________________

self = <test_asgi_ipc.IPCLayerTests testMethod=test_send_recv>

    def setUp(self):
        if self.channel_layer is None:
            raise ValueError("You must define 'channel_layer' when subclassing the conformance tests.")
        if self.expiry_delay is None:
            raise ValueError("You must define 'expiry_delay' when subclassing the conformance tests.")
        if "flush" in self.channel_layer.extensions:
>           self.channel_layer.flush()

/usr/lib/python2.7/site-packages/asgiref/conformance.py:43: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
asgi_ipc.py:164: in flush
    self.channel_store.flush()
asgi_ipc.py:299: in flush
    self._set_value(self.datatype())
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <asgi_ipc.MemoryDict object at 0x241ab10>, value = {}

    def _set_value(self, value):
        assert isinstance(value, self.datatype)
        try:
            self.semaphore.acquire(self.death_timeout)
        except posix_ipc.BusyError:
            self._reset()
            self.semaphore.acquire(0)
        try:
            self.mmap.seek(0)
            self.mmap.write(self.signature)
            towrite = msgpack.packb(value, use_bin_type=True)
>           self.mmap.write(struct.pack("!I", len(towrite)))
E           TypeError: Struct() argument 1 must be string, not unicode

asgi_ipc.py:265: TypeError
______________________________ IPCLayerTests.test_strings ______________________________

self = <test_asgi_ipc.IPCLayerTests testMethod=test_strings>

    def setUp(self):
        if self.channel_layer is None:
            raise ValueError("You must define 'channel_layer' when subclassing the conformance tests.")
        if self.expiry_delay is None:
            raise ValueError("You must define 'expiry_delay' when subclassing the conformance tests.")
        if "flush" in self.channel_layer.extensions:
>           self.channel_layer.flush()

/usr/lib/python2.7/site-packages/asgiref/conformance.py:43: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
asgi_ipc.py:164: in flush
    self.channel_store.flush()
asgi_ipc.py:299: in flush
    self._set_value(self.datatype())
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <asgi_ipc.MemoryDict object at 0x241ab10>, value = {}

    def _set_value(self, value):
        assert isinstance(value, self.datatype)
        try:
            self.semaphore.acquire(self.death_timeout)
        except posix_ipc.BusyError:
            self._reset()
            self.semaphore.acquire(0)
        try:
            self.mmap.seek(0)
            self.mmap.write(self.signature)
            towrite = msgpack.packb(value, use_bin_type=True)
>           self.mmap.write(struct.pack("!I", len(towrite)))
E           TypeError: Struct() argument 1 must be string, not unicode

asgi_ipc.py:265: TypeError
___________________________ MemoryDictTests.test_item_access ___________________________

self = <test_asgi_ipc.MemoryDictTests testMethod=test_item_access>

    def setUp(self):
        self.instance = MemoryDict("/test-md")
>       self.instance.flush()

test_asgi_ipc.py:21: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
asgi_ipc.py:299: in flush
    self._set_value(self.datatype())
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <asgi_ipc.MemoryDict object at 0x24de0d0>, value = {}

    def _set_value(self, value):
        assert isinstance(value, self.datatype)
        try:
            self.semaphore.acquire(self.death_timeout)
        except posix_ipc.BusyError:
            self._reset()
            self.semaphore.acquire(0)
        try:
            self.mmap.seek(0)
            self.mmap.write(self.signature)
            towrite = msgpack.packb(value, use_bin_type=True)
>           self.mmap.write(struct.pack("!I", len(towrite)))
E           TypeError: Struct() argument 1 must be string, not unicode

asgi_ipc.py:265: TypeError
======================== 15 failed, 14 skipped in 0.74 seconds =========================
error: Bad exit status from /var/tmp/rpm-tmp.tFY9zf (%check)

Outdate asgiref

It looks like this package is currently incompatible with the latest Django Channels as Channels requires asgiref 2.3.x and asgi_ipc requires asgiref 1.1.x.

Is there a real incompatibility here, or is it safe to bump the required asgiref version for this package?

Unresolved attribute reference `self.hosts`

See

asgi_ipc/asgi_ipc.py

Line 168 in b6750c5

return "%s(hosts=%s)" % (self.__class__.__name__, self.hosts)

I added a # TODO comment in my latest pull request.
980ccab#diff-64b224d30f0fb99ddead52f77c969d28R169

Sending messages to channels via celery

I'm experiencing a very similar problem to this one in my Django Development/Testing environment with the following configuration:

# settings.py
CHANNEL_LAYERS = {
    "default": {
        # This example app uses the Redis channel layer implementation
        # asgi_inmemory
        "BACKEND": "asgiref.inmemory.ChannelLayer",
        "ROUTING": "mytest.routing.channel_routing",
    },
}

My Problem: A scheduled (Celery) task can not send a web socket request through channels. However, it works great in views and other locations. The symptoms are present both with the immediately=True setting and without it. There is no error or anything; the web socket request is just simply not passed along. However having a signal generated from within a model/view works awesome (so i know my setup is good - or so i think).

# from within a celery task (this will just silently fail and not fire to the client):
Group("test").send({
    "text": json.dumps({
        'type': 'test',
    })
}, immediately=True)

# nothing gets sent to the user at this point

I'm launching a celery worker (before starting Django) like so:

   DJANGO_SETTINGS_MODULE=myapp.settings celery --app=myapp worker --beat \
      -l debug \
      --concurrency=1 \
      -Q celery,testqueueA,testqueueB \
      --pidfile for_killing_later.pid &

Was just curious if there was something i was missing here? My package versions are:

• Django==1.10.7
• channels==1.1.3
• asgi-redis==1.4.0
• asgi-ipc==1.4.0
• daphne==1.2.0
• asgiref==1.1.2
• celery==3.2.1

Any help and/or advice would be most welcome!

Move IPC layer to use SQLite in either a file or a shared memory segment

Rather than just dumping msgpack output into shared memory as it does right now.

Unable to use layer with python 3.6

First thought this is the same issue as in #26 as I got the same error in a docker container built on top of python:3.6.2-stretch, but also reproducible on a host machine. Using example from the readme:

$ pip3.6 install --user asgi_ipc
Collecting asgi_ipc
  Using cached asgi_ipc-1.4.0-py2.py3-none-any.whl
Collecting msgpack-python (from asgi_ipc)
Collecting asgiref~=1.1.2 (from asgi_ipc)
  Using cached asgiref-1.1.2-py2.py3-none-any.whl
Collecting posix-ipc>=1.0.0 (from asgi_ipc)
Requirement already satisfied: six in /usr/lib64/python3.6/site-packages (from asgi_ipc)
Installing collected packages: msgpack-python, asgiref, posix-ipc, asgi-ipc
Successfully installed asgi-ipc-1.4.0 asgiref-1.1.2 msgpack-python-0.4.8 posix-ipc-1.0.0
$ 
$ python3.6
Python 3.6.1 (default, Apr 26 2017, 12:37:57) 
[GCC 6.3.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import asgi_ipc
>>> import asgi_ipc as asgi
>>> channel_layer = asgi.IPCChannelLayer(
...     prefix="aeracode",
...     message_memory=200 * 1024 * 1024,
... )
>>> channel_layer.send("my_channel", {"text": "Hello ASGI"})
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/hoefling/.local/lib64/python3.6/site-packages/asgi_ipc/core.py", line 61, in send
    channel_size = self.message_store.length(channel)
  File "/home/hoefling/.local/lib64/python3.6/site-packages/asgi_ipc/store.py", line 134, in length
    with self.mutate_value() as value:
  File "/usr/lib64/python3.6/contextlib.py", line 82, in __enter__
    return next(self.gen)
  File "/home/hoefling/.local/lib64/python3.6/site-packages/asgi_ipc/store.py", line 50, in mutate_value
    value = pickle.load(self.mmap)
_pickle.UnpicklingError: invalid load key, '\x00'.
>>>

Not reproducible with python 3.5:

$ python3.5
Python 3.5.3 (default, Mar 25 2017, 06:59:48) 
[GCC 6.3.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import asgi_ipc as asgi
>>> channel_layer = asgi.IPCChannelLayer(
...     prefix="aeracode",
...     message_memory=200 * 1024 * 1024,
... )
>>> channel_layer.send("my_channel", {"text": "Hello ASGI"})
>>> exit()

shared memory segment and semaphore names suffixes are too large for OS X

Asgi_ipc uses "/%s-channel-dict" for the shared memory segment name and "/%s-channel-dict-semaphore" (where %s is the configured prefix).

Mac OS X's IPC name sizes cannot be more than 30 bytes (source, second list item). This means that the maximum usable prefix on OS X is 6 characters, which is pretty small.

One solution would be to change the suffixes to something smaller (like "-chan" and "-chan-sem"). I can provide a PR if this solution is ok.

Thanks.

MemoryError on large uploads

Looks like Daphne will choke on large POST requests (in my case while uploading a large file).

out 16 16:48:12 icr daphne[29020]: 2017-10-16 16:48:12,424 ERROR    Traceback (most recent call last):
out 16 16:48:12 icr daphne[29020]:   File "/var/app/virtualenvs/ev/lib/python3.6/site-packages/daphne/http_protocol.py", line 174, in process
out 16 16:48:12 icr daphne[29020]:     "server": self.server_addr,
out 16 16:48:12 icr daphne[29020]:   File "/var/app/virtualenvs/ev/lib/python3.6/site-packages/asgi_ipc/core.py", line 68, in send
out 16 16:48:12 icr daphne[29020]:     time.time() + self.expiry,
out 16 16:48:12 icr daphne[29020]:   File "/var/app/virtualenvs/ev/lib/python3.6/site-packages/asgi_ipc/store.py", line 116, in append
out 16 16:48:12 icr daphne[29020]:     value.setdefault(name, []).append((item, expiry))
out 16 16:48:12 icr daphne[29020]:   File "/usr/lib/python3.6/contextlib.py", line 89, in __exit__
out 16 16:48:12 icr daphne[29020]:     next(self.gen)
out 16 16:48:12 icr daphne[29020]:   File "/var/app/virtualenvs/ev/lib/python3.6/site-packages/asgi_ipc/store.py", line 64, in mutate_value
out 16 16:48:12 icr daphne[29020]:     pickle.dump(value, self.mmap, protocol=2)
out 16 16:48:12 icr daphne[29020]: MemoryError

I'm using Python 3.6.1 (64bit) on Ubuntu 17.04 with 8G RAM. Uploaded file is 1.6GB.

Looks like a problem with pickle, is there any way to use other pickle-compatible serializer that is better for larger objects?

Semaphores don't get created on Alpine Linux

Using Alpine Linux 3.6, semaphores don't seem to be getting created in /dev/shm:

# ls -l /dev/shm
total 8
-rw-------    1 root     root      20971520 Oct 18 15:34 asgi-groups
-rw-------    1 root     root     104857600 Oct 18 15:34 asgi-messages

Using the same versions listed below on Debian Jessie, the semaphores and shared memory appear in /dev/shm:

# ls -l /dev/shm
total 12
-rw------- 1 root root  20971520 Oct 18 15:38 asgi-groups
-rw------- 1 root root 104857600 Oct 18 15:38 asgi-messages
-rw------- 1 root root        32 Oct 18 15:38 sem.asgi-groups
-rw------- 1 root root        32 Oct 18 15:38 sem.asgi-messages

I'm not sure if it's related to the missing semaphore, but the generated value ends up being None. In Debian, I can at least see the value as {}.

Once daphne or a second worker are started, then the value can't be unpickled and results in the stack trace below.

---

On Alpine:

Python 3.6.1 (default, May  2 2017, 15:16:41)
[GCC 6.3.0] on linux

On Debian:

Python 3.6.3 (default, Oct 10 2017, 02:29:16)
[GCC 4.9.2] on linux

Both:

asgi-ipc==1.4.1
asgiref==1.1.
channels==1.1.8
daphne==1.3.0
Django==1.11.6
posix-ipc==1.0.0

From Alpine:

  File "./manage.py", line 22, in <module>
    execute_from_command_line(sys.argv)
  File "/usr/lib/python3.6/site-packages/django/core/management/__init__.py", line 364, in execute_from_command_line
    utility.execute()
  File "/usr/lib/python3.6/site-packages/django/core/management/__init__.py", line 356, in execute
    self.fetch_command(subcommand).run_from_argv(self.argv)
  File "/usr/lib/python3.6/site-packages/django/core/management/base.py", line 283, in run_from_argv
    self.execute(*args, **cmd_options)
  File "/usr/lib/python3.6/site-packages/django/core/management/base.py", line 330, in execute
    output = self.handle(*args, **options)
  File "/usr/lib/python3.6/site-packages/channels/management/commands/runworker.py", line 83, in handle
    worker.run()
  File "/usr/lib/python3.6/site-packages/channels/worker.py", line 87, in run
    channel, content = self.channel_layer.receive_many(channels, block=True)
  File "/usr/lib/python3.6/site-packages/asgiref/base_layer.py", line 43, in receive_many
    return self.receive(channels, block)
  File "/usr/lib/python3.6/site-packages/asgi_ipc/core.py", line 83, in receive
    message = self.message_store.pop(channel)
  File "/usr/lib/python3.6/site-packages/asgi_ipc/store.py", line 123, in pop
    with self.mutate_value() as value:
  File "/usr/lib/python3.6/contextlib.py", line 82, in __enter__
    return next(self.gen)
  File "/usr/lib/python3.6/site-packages/asgi_ipc/store.py", line 57, in mutate_value
    value = pickle.load(self.mmap)
_pickle.UnpicklingError: invalid load key, '\xff'

Don't use pkg_resources to get version number

This doesn't work in older version of setuptools.
Maybe set it both in setup.py and asgi_ipc.py ?

MemoryDatastructure and double acquisition of semaphores

Hello,

asgi_ipc.py:MemoryDatastructure._get_value() and MemoryDatastructure._set_value() acquire a semaphore, read/modify the shared memory and eventually release the semaphore.

The methods MemoryDatastructure.__setitem__, MemoryDatastructure __delitem__ and MemoryDict.discard call first _get_value() and then _set_value. This means, that they acquire the semaphore, read the data, release the semaphore, modify the data, acquire the semaphore again, write the new data, and release the semaphore.

Doesn't this cause race conditions?

Wouldn't asgi_ipc get faster, if during MemoryDatastructre.__setitem__ the semaphore is acquired and released only once?

Suggestion: Raise a ValueError exception if receive()'s "channels" argument is not of type "list".

Yes, I should have read the documentation more carefully, especially the part where it says

receive(channels, block=False), a callable that takes a list of channel names as unicode strings […]*

It was also tempting to write

channel_layer.send("my_channel", {"text": "Hello"})
channel_layer.receive("my_channel")

and pass a string to the receive method instead of a list, just as in the send() method

Took me a moment to figure out why I wouldn't receive any messages. In asgi_ipc.py#L67 the channels argument is passed to list() and therefore I was listening to a number of channels, but not to what I thought I were listening.

My suggestion is to raise a ValueError exception if the channels argument is not of type "list". Makes debugging a little quicker for those who happen to have overlooked that part in the docs.

* Source: https://channels.readthedocs.io/en/latest/asgi.html

Running channels as non-root user gives "posix_ipc.PermissionsError: Permission denied"

I am getting "posix_ipc.PermissionsError: Permission denied" when running channels example as non-root user:

$ python3 manage.py runserver
Unhandled exception in thread started by <function check_errors.<locals>.wrapper at 0x7f693358e158>
Traceback (most recent call last):
  File "/usr/local/lib/python3.5/dist-packages/django/utils/autoreload.py", line 226, in wrapper
    fn(*args, **kwargs)
  File "/usr/local/lib/python3.5/dist-packages/channels/management/commands/runserver.py", line 38, in inner_run
    self.channel_layer = channel_layers[DEFAULT_CHANNEL_LAYER]
  File "/usr/local/lib/python3.5/dist-packages/channels/asgi.py", line 53, in __getitem__
    self.backends[key] = self.make_backend(key)
  File "/usr/local/lib/python3.5/dist-packages/channels/asgi.py", line 44, in make_backend
    asgi_layer = backend_class(**self.configs[name].get("CONFIG", {}))
  File "/usr/local/lib/python3.5/dist-packages/asgi_ipc.py", line 43, in __init__
    self.channel_store = MemoryDict("/%s-chan" % self.prefix, size=channel_memory)
  File "/usr/local/lib/python3.5/dist-packages/asgi_ipc.py", line 201, in __init__
    initial_value=1,
posix_ipc.PermissionsError: Permission denied
^CException ignored in: <bound method MemoryDatastructure.__del__ of <asgi_ipc.MemoryDict object at 0x7f6933590630>

It tries to use /asgi-chan path for Semaphore and promptly fails. This error doesn't happen when running as root user. Please see line 43 and 45 in asgi_ipc.py