Full in memory channel layer #863

svenauhagen · 2018-02-05T14:01:12Z

Hi,

I added the full in memory layer back to channels 2.
We always used it in channels 1 for local testing.

Can you merge it into the code?

Best and thanks
Sven

andrewgodwin · 2018-02-05T18:26:22Z

I will happily take this once it has unit tests (I'll write the documentation part) - you may want to reuse the channels_redis unit tests for the most part.

svenauhagen · 2018-02-05T18:48:52Z

I ported over the channel_redis tests to the inmemory layer.
Let me know if that is alright or if you want me to add additional tests.

andrewgodwin

Having looked over the code more, I'm not sure you need all the complexity that the Redis client has with receiving loops - you should be able to use asyncio Queue objects for most of the operations here and await on them natively (and there's no need to pack and unpack around things with ! in them as that's purely a network efficiency thing).

Do you think you'd be able to reduce it down to have less of the receive_loop stuff? The version you're replacing was a lot simpler.

svenauhagen · 2018-02-06T09:46:39Z

Ah yes, I wasn't aware of the asyncio queue.
I am knew to asyncio just learned it over the weekend to write the memory channel layer :)

Do I understand you correctly that I should simply use the raw channel name for the queue key?
Should I also dismiss the assert self.valid_channel_name(channel, receive=True) check then?

andrewgodwin · 2018-02-06T17:16:33Z

You can leave the valid check in, but there's no need to do the part where it takes a name with a ! in it and splits it into a non-local part, and puts the real channel name into the message dict.

svenauhagen · 2018-02-06T17:33:18Z

Alright, can you look at it now.
I tested it today and it worked well that way.

brianmay · 2018-02-07T01:25:39Z

channels/layers.py

-    In-memory channel layer implementation for testing purposes.
-    """
+    '''
+    Our own in memory layer


Did you mean to change the docstring?

No that was left in there from testing I changed it to "In-memory channel layer implementation" since it can be used for more than testing now

andrewgodwin · 2018-02-07T07:29:51Z

channels/layers.py

+        queue = self.channels.setdefault(channel, asyncio.Queue())
+
+        # Do a plain direct receive
+        _, message = await queue.get()


I would add something here to clean up the queue entry in the self.channels dict if it's now empty.

andrewgodwin · 2018-02-07T07:30:34Z

channels/layers.py

-    """
+    '''
+    In-memory channel layer implementation
+    '''


Please change these back to """ - trying to keep everything consistent!

andrewgodwin · 2018-02-07T07:31:30Z

channels/layers.py

        self.channels = {}
+        self.groups = {}
+        self.thread_lock = threading.Lock()


I'm not sure a threading.Lock makes sense here - you shouldn't need it since you're writing async code. Did you find a reason to add it?

andrewgodwin · 2018-02-07T07:32:23Z

channels/layers.py

+                    self._remove_from_groups(channel)
+                # Is the channel now empty and needs deleting?
+                if not queue:
+                    del self.channels[channel]


This method should also clean expired group memberships separately.

What do you mean by that exactly?
Should I remove the channel from all groups if not queue?

You need to go through the group memberships and expire the entries if their expiry time is bigger than group_expiry (there should be a test for this too, but I'll add that later). The Redis one does it here: https://github.com/django/channels_redis/blob/master/channels_redis/core.py#L316

andrewgodwin

Looking good - just a few small cleanups, but this is getting close to merge! Thanks for your work so far.

svenauhagen · 2018-02-07T07:37:50Z

Great, I will get on these cleanups later today.
Regarding the thread lock I do have a general question since I am not sure.
How is daphne handling new web requests in 2.0?
Is it spawning a thread for each request or is it entirely done with asyncio coroutines?

andrewgodwin · 2018-02-07T07:39:40Z

It's all coroutines. Some requests will end up using threads if they're based on SyncConsumer, but all the async stuff (calls to channel layers included) happens in a single main thread on a single event loop.

svenauhagen · 2018-02-07T07:44:05Z

Thanks for the explanation.
Isn't that a problem though if some async requests run longer without await daphne will be blocked to answer new requests?

andrewgodwin · 2018-02-07T07:56:09Z

No, because they're async requests, as long as they await anything that's long running (as everything is, like receiving off the channel layer) the event loop frees up. You can have thousands of coroutines in the same event loop easily.

svenauhagen · 2018-02-08T11:06:33Z

Ok I think I got it now

proofit404 · 2018-02-08T11:57:49Z

channels/layers.py

@@ -267,7 +266,8 @@ def _clean_expired(self):
        for group in self.groups:
            for channel in self.groups.get(group, set()):
                # If join time is older than group_expiry end the group membership
-                if self.groups[group][channel] and int(self.groups[group][channel]) < (int(time.time()) - self.group_expiry):
+                if (self.groups[group][channel] and
+                    int(self.groups[group][channel]) < (int(time.time()) - self.group_expiry)):
                    del self.groups[group][channel]


Maybe shortcut local variable will be better.

I added a timeout variable

andrewgodwin · 2018-02-09T23:54:09Z

Looks good! I am ready to merge this if you are,

brubbel · 2018-02-24T18:40:24Z

@svenauhagen @andrewgodwin

channels/channels/layers.py

Lines 268 to 272 in d6643d9

    
           for channel in self.groups.get(group, set()): 
        
               # If join time is older than group_expiry end the group membership 
        
               if self.groups[group][channel] and int(self.groups[group][channel]) < timeout: 
        
                   # Delete from group 
        
                   del self.groups[group][channel]

In python3, looping over a dict/set defaults to iterators, and _clean_expired() raises an exception, as changing the dict/set is not allowed during iteration.
for channel in self.groups.get(group, set()):
should be
for channel in list(self.groups.get(group, set())):

Error below:

Exception in thread Thread-5:
Traceback (most recent call last):
  File "/usr/lib/python3.5/threading.py", line 914, in _bootstrap_inner
    self.run()
  File "/usr/lib/python3.5/threading.py", line 862, in run
    self._target(*self._args, **self._kwargs)
  File "/home/xyz/venv3/lib/python3.5/site-packages/asgiref/sync.py", line 49, in __call__
    return call_result.result()
  File "/usr/lib/python3.5/concurrent/futures/_base.py", line 405, in result
    return self.__get_result()
  File "/usr/lib/python3.5/concurrent/futures/_base.py", line 357, in __get_result
    raise self._exception
  File "/home/xyz/venv3/lib/python3.5/site-packages/asgiref/sync.py", line 63, in main_wrap
    result = await self.awaitable(*args, **kwargs)
  File "/home/xyz/django/abc/consumers.py", line 47, in sender
    await self.channel_layer.group_send(self.MY_GROUP, message)
  File "/home/xyz/venv3/lib/python3.5/site-packages/channels/channels/layers.py", line 321, in group_send
    self._clean_expired()
  File "/home/xyz/venv3/lib/python3.5/site-packages/channels/channels/layers.py", line 268, in _clean_expired
    for channel in self.groups.get(group, set()):
RuntimeError: dictionary changed size during iteration

andrewgodwin · 2018-02-24T18:42:52Z

I just pushed up a commit to fix that.

brubbel · 2018-02-25T09:45:15Z

I would like to point out some additional issues.

channels/channels/layers.py

Lines 261 to 263 in b58c213

    
           # Is the channel now empty and needs deleting? 
        
           if not queue: 
        
               del self.channels[channel]

Previously (before asyncio), self.channels had been using a collections.deque, for which the following is valid to check if queue is empty:

if not queue:
     do_something()

However, the truth value of a queue.Queue or asyncio.Queue cannot be used to check whether the queue is empty. It will always return True and the channel is never deleted. The correct statement is as follows:

if queue.empty():
     do_something()

With the above corrected statement effectively purging channels from the self.channels dict, it also starts to delete non-expired (empty, but still active) channels. This breaks InMemoryChannelLayer again.
The corrected statement, which only purges expired channels, is as follows:

if remove and queue.empty():
    del self.channels[channel]

brubbel · 2018-02-25T14:43:17Z

Even with the above fixes, the InMemoryChannelLayer keeps leaking channel objects in self.channels {}, when websocket connections are coming and going over time. The group_discard() method is called, which is correct, but channels with an empty queue keep hanging around.

I suggest the following fixes to _group_expire(), please comment if I missed something:
brubbel@ad6b135

andrewgodwin · 2018-02-25T18:24:48Z

Can you open a separate issue to track this? It's getting complex enough it needs one.

It's also worth noting that the in-memory channel layer is really only meant for testing, so fixes to its performance/longevity are going to be lower priority than some other bugs in the queue!

brubbel · 2018-02-25T18:41:36Z

I am fully aware that the in-memory channel layer is for debug only, however -imho- for applications on a lightweight system such as raspberry pi, I think it is a reasonable alternative.

andrewgodwin · 2018-02-25T18:49:58Z

Right, just saying that fixing this is going to come below the 6 or so other bugs I have to look at at the moment!

svenauhagen added 2 commits February 5, 2018 14:48

in memory channel layer

8db8ad6

in memory layer

5275e71

svenauhagen changed the title ~~Fill in memory channel layer~~ Full in memory channel layer Feb 5, 2018

svenauhagen added 3 commits February 5, 2018 15:08

style fixes

a0323c6

style fixes

039c96d

style fixes

f130dfb

svenauhagen added 3 commits February 5, 2018 19:34

inmemory layer tests

c3d1339

style fix

bffaef5

test requirements

4d47f87

andrewgodwin reviewed Feb 6, 2018

View reviewed changes

svenauhagen added 4 commits February 6, 2018 18:18

User asyncio queue

1b87609

Remove loop line

bbe5fc7

style fix and remove old tests

3f4b63b

style fix and remove old tests

e75a422

brianmay reviewed Feb 7, 2018

View reviewed changes

Fix docstring

e16c38f

andrewgodwin reviewed Feb 7, 2018

View reviewed changes

changes

52d31ee

Merge branch 'master' of github.com:svenauhagen/channels

92800c9

andrewgodwin mentioned this pull request Feb 7, 2018

Issues with InMemoryChannelLayer #882

Closed

svenauhagen added 4 commits February 8, 2018 11:18

group expiration

5700171

style fixes

5d81d45

style fixes

dc93e6d

style fixes

9c8f8a1

proofit404 reviewed Feb 8, 2018

View reviewed changes

local timeout variable in group expiry

6286fd2

andrewgodwin approved these changes Feb 9, 2018

View reviewed changes

orn688 mentioned this pull request Feb 13, 2018

Add WebsocketConsumer group join/leave functionality #901

Merged

andrewgodwin merged commit d65ffe7 into django:master Feb 13, 2018

brubbel mentioned this pull request Feb 25, 2018

InMemoryChannelLayer leaks orphaned channels #933

Open

nicksellen mentioned this pull request Mar 12, 2018

WIP: Upgrade to django channels 2 karrot-dev/karrot-backend#532

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Full in memory channel layer #863

Full in memory channel layer #863

svenauhagen commented Feb 5, 2018

andrewgodwin commented Feb 5, 2018

svenauhagen commented Feb 5, 2018

andrewgodwin left a comment

svenauhagen commented Feb 6, 2018

andrewgodwin commented Feb 6, 2018

svenauhagen commented Feb 6, 2018

brianmay Feb 7, 2018

svenauhagen Feb 7, 2018

andrewgodwin Feb 7, 2018

andrewgodwin Feb 7, 2018

andrewgodwin Feb 7, 2018

andrewgodwin Feb 7, 2018

svenauhagen Feb 7, 2018

andrewgodwin Feb 7, 2018

andrewgodwin left a comment

svenauhagen commented Feb 7, 2018

andrewgodwin commented Feb 7, 2018

svenauhagen commented Feb 7, 2018

andrewgodwin commented Feb 7, 2018

svenauhagen commented Feb 8, 2018

proofit404 Feb 8, 2018

svenauhagen Feb 8, 2018

andrewgodwin commented Feb 9, 2018

brubbel commented Feb 24, 2018

andrewgodwin commented Feb 24, 2018

brubbel commented Feb 25, 2018 •

edited

brubbel commented Feb 25, 2018 •

edited

andrewgodwin commented Feb 25, 2018 •

edited

brubbel commented Feb 25, 2018

andrewgodwin commented Feb 25, 2018

Full in memory channel layer #863

Full in memory channel layer #863

Conversation

svenauhagen commented Feb 5, 2018

andrewgodwin commented Feb 5, 2018

svenauhagen commented Feb 5, 2018

andrewgodwin left a comment

Choose a reason for hiding this comment

svenauhagen commented Feb 6, 2018

andrewgodwin commented Feb 6, 2018

svenauhagen commented Feb 6, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

andrewgodwin left a comment

Choose a reason for hiding this comment

svenauhagen commented Feb 7, 2018

andrewgodwin commented Feb 7, 2018

svenauhagen commented Feb 7, 2018

andrewgodwin commented Feb 7, 2018

svenauhagen commented Feb 8, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

andrewgodwin commented Feb 9, 2018

brubbel commented Feb 24, 2018

andrewgodwin commented Feb 24, 2018

brubbel commented Feb 25, 2018 • edited

brubbel commented Feb 25, 2018 • edited

andrewgodwin commented Feb 25, 2018 • edited

brubbel commented Feb 25, 2018

andrewgodwin commented Feb 25, 2018

brubbel commented Feb 25, 2018 •

edited

brubbel commented Feb 25, 2018 •

edited

andrewgodwin commented Feb 25, 2018 •

edited