Reimplementation of offline subscriptions #31

behrad · 2017-07-03T11:50:28Z

dropped counter:offline:subscriptions key. offline subs are SCARD of subs SET
dropped counter:offline:clients key. offline clients are SCARD of clients SET
dropped sub:client, no subscriptions index of clientIds are stored any more. all subscription list api methods are implemented VIA internal Qlobber.
renamed client:sub: to client:
new clients SET added containing all client ids
new subs SET added containing all topics
all redis operations are handled by _getPipeline

Would you please review these @mcollina @GavinDmello ?

- switch all redis operations to _getPipeline

behrad · 2017-07-03T11:54:52Z

tests are failing 'cause of this issue
which this implementation recovered. It will pass if you merge that in @mcollina

mcollina · 2017-07-03T12:03:17Z

persistence.js

+//   this.clientId = clientId
+//   this.topic = topic
+//   this.qos = qos
+// }


Can you please remove this if it is not needed anymore?

mcollina · 2017-07-03T12:04:52Z

Do you have a benchmark result for this?

behrad · 2017-07-03T12:15:43Z

Do you have a benchmark result for this?

Doing this right now... Will post results.

behrad · 2017-07-03T19:39:32Z

Implementation	offline clients	subscriptions count	startup time	heap	rss	Redis Memory
Current	1384138	2879070	406500.858ms	2423.3MB	3552.9MB	2.5G
This PR	1384066	2878969	245929.751ms	2265.2MB	2798.9MB	864M

behrad · 2017-07-03T19:49:35Z

a single Aedes process is able to handle 250 connections each with 3 subscriptions per second (around 700 subs/sec) in my tests, but any thing less or near eats cpu above 90%,
for this part I should check aedes-cached-persistence and aedes

mcollina · 2017-07-03T20:29:48Z

These are impressive numbers. Definitely +1. @GavinDmello what do you think?

GavinDmello · 2017-07-04T05:47:26Z

@mcollina @behrad This looks good. I have some suggestions though

GavinDmello · 2017-07-03T12:40:39Z

persistence.js

-  multi.exec(cb)
+  var pipeline = this._getPipeline()
+  pipeline.rpush(listKey, key)
+  pipeline.set(key, msgpack.encode(new Packet(packet)), cb)


Can we simply use db.set here ? These keys can be sharded in case of nutcracker. It won't be sharded this case.

I think we may revert all _getPipeline usages to direct _db to support redis cluster/nutcracker

GavinDmello · 2017-07-03T12:41:10Z

persistence.js

-
-  multi.exec(cb)
+  var pipeline = this._getPipeline()
+  pipeline.rpush(listKey, key)


I think this can be a Redis set too for the sake of dedupe

Can you explain a scenario we may need deduping on packet.brokerId + ':' + packet.brokerCounter inside outgoing messages for a client @GavinDmello ?

This is just to be safe. Using a list was a bad idea here, I guess.

by SMEMBERS we can't control the offline message flow to the client.
lets see what @mcollina thinks about this?

I think the list is the correct behavior, as we to retrieve the elements in insertion order.

Alright 👍

behrad

good news: _db is performing better than _getPipeline in _setup where we are doing heavy hgetalls.

behrad · 2017-07-04T05:54:41Z

persistence.js

-  multi.exec(cb)
+  var pipeline = this._getPipeline()
+  pipeline.rpush(listKey, key)
+  pipeline.set(key, msgpack.encode(new Packet(packet)), cb)


I think we may revert all _getPipeline usages to direct _db to support redis cluster/nutcracker

behrad · 2017-07-06T15:09:48Z

implemented createRetainedStreamCombi here & upgraded to aedes-cached-persistence 4.0.0 @mcollina

behrad · 2017-07-07T07:09:38Z

@mcollina ping

mcollina

Can you remove completely the pipeline support? It is mostly gone now, so we can completely remove it then.

mcollina · 2017-07-07T07:56:12Z

persistence.js

      count++
-      that._waitFor(client, sub.topic, finish)
+      // TODO don't wait the client an extra tick


Despite mosca, aedes is publishing sub topics, and even the local process is waiting for the subscribed handler (inside cached-persistence). My points:

to call the callback sooner, we may go mosca way and check broker.id inside message handler. Or we may even call the callback without this check and let the subscription take effect in background after client subacked !?

for subscription publication between nodes, cached-persistence is using broker's pub/sub. Could this be handled any lower level, not to pollute MQTT pub/sub with inter-node messaging!?

this is needed to provide a stable unit testing base. Mosca had so many spurious unit tests, and some of them relies on setTimeout to work. One of the underlining goal of aedes is to make maintenance and evolution simpler, and this goes into that direction.

no

I'm ok if you would like to try and remove that (or maybe disable it with an option?), but let's drop the TODO for now.

removed TODO 👍

mcollina · 2017-07-07T07:57:08Z

persistence.js

+
+  patterns.map(function (pattern) {
+    qlobber.add(pattern, true)
+  })


you should use a for loop here, it's way faster is this is going to be in a hot path.

mcollina · 2017-07-07T07:57:22Z

persistence.js

 function RedisPersistence (opts) {
  if (!(this instanceof RedisPersistence)) {
    return new RedisPersistence(opts)
  }

+  opts = opts || {}
+  this.maxSessionDelivery = opts.maxSessionDelivery || 1000


Can you please document this option?

mcollina · 2017-07-07T07:59:19Z

persistence.js

  }

+  this._db.srem(subsKey, subs) // TODO matcher.match should be checked
+
+  this._db.hdel(clientSubKey, subs)


we should specify callbacks for this, and deal with potential errors.

can you check this now @mcollina ?

mcollina · 2017-07-07T08:02:47Z

persistence.js

    if (err) {
      splitStream.emit('error', err)
    } else {
-      splitStream.write(results)
+      splitStream.write(clientIds)


You can completely avoid the use of splitStream here, just do the processing and call write on the throughv instance

mcollina

LGTM but I would like @GavinDmello review as well.

behrad · 2017-07-08T08:38:58Z

@GavinDmello ping

GavinDmello · 2017-07-08T09:21:32Z

persistence.js

-  var pipeline = this._getPipeline()
-  pipeline.lrem(willKey, 0, key)
-  pipeline.getBuffer(key, function getClientWill (err, packet) {
+  var that = this


That is not required here.

why not? that is used in line 479

is this ok @GavinDmello ?

I meant we can just use this as its in the same scope :)

I am using that in 479 inside a callback, Am I missing something?

You mean the getBuffer callback? This is weird, it shows me it's in the same scope. i.e 479 is outside the getClientWill function.

Yes, del should have been inside the getBuffer after multi being removed, fixed it just now @GavinDmello

GavinDmello · 2017-07-08T11:05:57Z

Minor change I mentioned above. I'm 👍 with this PR .

behrad · 2017-07-09T06:34:31Z

I'm not sure why node v5 is failed!

GavinDmello · 2017-07-09T06:43:20Z

Let me try running it again

- re-implementation of offline client subscriptions

0adb302

- switch all redis operations to _getPipeline

behrad mentioned this pull request Jul 3, 2017

storing QOS=0 subscriptions? #30

Closed

This was referenced Jul 3, 2017

Unsubscribing a topic is removing all topic subscriptions from Qlobber moscajs/aedes-cached-persistence#4

Merged

processKeysForClient should not be working #29

Closed

mcollina requested a review from GavinDmello July 3, 2017 12:02

mcollina reviewed Jul 3, 2017

View reviewed changes

behrad added 3 commits July 3, 2017 16:59

remove comments

fc7500c

fix setup

9a5719f

fix _setup

0c47ed3

GavinDmello reviewed Jul 4, 2017

View reviewed changes

behrad commented Jul 4, 2017

View reviewed changes

behrad added 2 commits July 4, 2017 12:19

revert all redis operations to _db

e52584c

upgrade aedes-cached-persistence, implement createRetainedStreamCombi

2470915

mcollina reviewed Jul 7, 2017

View reviewed changes

behrad added 2 commits July 7, 2017 13:59

review changes

18774c6

removed TODO

9cf7eeb

mcollina approved these changes Jul 7, 2017

View reviewed changes

GavinDmello reviewed Jul 8, 2017

View reviewed changes

behrad mentioned this pull request Jul 8, 2017

Fix invalid subscription handling moscajs/aedes#121

Merged

move del inside getBuffer

759a394

behrad mentioned this pull request Jul 9, 2017

Using aedes-persistence-redis with Aedes resulting in slow client connections. #10

Open

mcollina merged commit dce519b into moscajs:master Jul 10, 2017

behrad deleted the next_version branch July 11, 2017 15:14

behrad mentioned this pull request Jul 12, 2017

startup is limited to 10000 clients #28

Closed

Reimplementation of offline subscriptions #31

Reimplementation of offline subscriptions #31

Conversation

behrad commented Jul 3, 2017

behrad commented Jul 3, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mcollina commented Jul 3, 2017

behrad commented Jul 3, 2017 • edited Loading

behrad commented Jul 3, 2017 • edited Loading

behrad commented Jul 3, 2017

mcollina commented Jul 3, 2017

GavinDmello commented Jul 4, 2017 • edited Loading

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

GavinDmello Jul 4, 2017 • edited Loading

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

behrad left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

behrad commented Jul 6, 2017

behrad commented Jul 7, 2017

mcollina left a 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

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

mcollina left a comment

Choose a reason for hiding this comment

behrad commented Jul 8, 2017

Choose a reason for hiding this comment

behrad Jul 8, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

behrad Jul 9, 2017 • edited Loading

Choose a reason for hiding this comment

GavinDmello Jul 9, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

GavinDmello commented Jul 8, 2017

behrad commented Jul 9, 2017

GavinDmello commented Jul 9, 2017

behrad commented Jul 3, 2017 •

edited

Loading

behrad commented Jul 3, 2017 •

edited

Loading

GavinDmello commented Jul 4, 2017 •

edited

Loading

GavinDmello Jul 4, 2017 •

edited

Loading

behrad Jul 8, 2017 •

edited

Loading

behrad Jul 9, 2017 •

edited

Loading

GavinDmello Jul 9, 2017 •

edited

Loading