From 882464f941da1df59c5fec44970427318ed56bc4 Mon Sep 17 00:00:00 2001
From: Yao Yue <yao@twitter.com>
Date: Tue, 4 Apr 2017 00:11:03 -0700
Subject: [PATCH 1/9] add data structures for topics, subscribers

---
 src/protocol/data/redis/cmd_pubsub.h |   7 +
 src/protocol/data/redis/request.h    |   2 +
 src/storage/CMakeLists.txt           |   1 +
 src/storage/pubsub/CMakeLists.txt    |   5 +
 src/storage/pubsub/README            |   5 +
 src/storage/pubsub/hashtable.h       |   4 +
 src/storage/pubsub/index.h           |  10 ++
 src/storage/pubsub/listener.c        | 218 +++++++++++++++++++++++++++
 src/storage/pubsub/listener.h        |  41 +++++
 src/storage/pubsub/topic.c           | 218 +++++++++++++++++++++++++++
 src/storage/pubsub/topic.h           |  41 +++++
 11 files changed, 552 insertions(+)
 create mode 100644 src/protocol/data/redis/cmd_pubsub.h
 create mode 100644 src/storage/pubsub/CMakeLists.txt
 create mode 100644 src/storage/pubsub/README
 create mode 100644 src/storage/pubsub/hashtable.h
 create mode 100644 src/storage/pubsub/index.h
 create mode 100644 src/storage/pubsub/listener.c
 create mode 100644 src/storage/pubsub/listener.h
 create mode 100644 src/storage/pubsub/topic.c
 create mode 100644 src/storage/pubsub/topic.h

diff --git a/src/protocol/data/redis/cmd_pubsub.h b/src/protocol/data/redis/cmd_pubsub.h
new file mode 100644
index 000000000..446cf4ec7
--- /dev/null
+++ b/src/protocol/data/redis/cmd_pubsub.h
@@ -0,0 +1,7 @@
+#pragma once
+
+/*          type                string          # of args */
+#define REQ_PUBSUB(ACTION)                            \
+    ACTION( REQ_PUBLISH,        "publish",      3  )\
+    ACTION( REQ_SUBSCRIBE,      "subscribe",    -2 )\
+    ACTION( REQ_UNSUBSCRIBE,    "unsubscribe",  -2 )
diff --git a/src/protocol/data/redis/request.h b/src/protocol/data/redis/request.h
index d5fb5f938..4289bd344 100644
--- a/src/protocol/data/redis/request.h
+++ b/src/protocol/data/redis/request.h
@@ -2,6 +2,7 @@
 
 #include "cmd_hash.h"
 #include "cmd_misc.h"
+#include "cmd_pubsub.h"
 #include "cmd_zset.h"
 
 #include <cc_array.h>
@@ -44,6 +45,7 @@ typedef enum cmd_type {
     REQ_HASH(GET_TYPE)
     REQ_ZSET(GET_TYPE)
     REQ_MISC(GET_TYPE)
+    REQ_PUBSUB(GET_TYPE)
     REQ_SENTINEL
 } cmd_type_e;
 #undef GET_TYPE
diff --git a/src/storage/CMakeLists.txt b/src/storage/CMakeLists.txt
index c06eede35..0756902af 100644
--- a/src/storage/CMakeLists.txt
+++ b/src/storage/CMakeLists.txt
@@ -1,2 +1,3 @@
 add_subdirectory(cuckoo)
+add_subdirectory(pubsub)
 add_subdirectory(slab)
diff --git a/src/storage/pubsub/CMakeLists.txt b/src/storage/pubsub/CMakeLists.txt
new file mode 100644
index 000000000..4578d1360
--- /dev/null
+++ b/src/storage/pubsub/CMakeLists.txt
@@ -0,0 +1,5 @@
+set(SOURCE
+    listener.c
+    topic.c)
+
+add_library(pubsub ${SOURCE})
diff --git a/src/storage/pubsub/README b/src/storage/pubsub/README
new file mode 100644
index 000000000..7def09c6a
--- /dev/null
+++ b/src/storage/pubsub/README
@@ -0,0 +1,5 @@
+Mapping #1: client -> channel, to clean-up when client quits
+Mapping #2: channel -> client, to handle subscription, fanout
+
+a channel with no subscriber should be deleted.
+a client with no channel subscription is kept.
diff --git a/src/storage/pubsub/hashtable.h b/src/storage/pubsub/hashtable.h
new file mode 100644
index 000000000..8de665265
--- /dev/null
+++ b/src/storage/pubsub/hashtable.h
@@ -0,0 +1,4 @@
+#pragma once
+
+#define HASHSIZE(_n) (1ULL << (_n))
+#define HASHMASK(_n) (HASHSIZE(_n) - 1)
diff --git a/src/storage/pubsub/index.h b/src/storage/pubsub/index.h
new file mode 100644
index 000000000..f9c8bc71a
--- /dev/null
+++ b/src/storage/pubsub/index.h
@@ -0,0 +1,10 @@
+#pragma once
+
+#include <cc_queue.h>
+
+struct index_node {
+    TAILQ_ENTRY(index_node) i_tqe;
+    void                    *obj;
+};
+
+TAILQ_HEAD(index_tqh, index_node);
diff --git a/src/storage/pubsub/listener.c b/src/storage/pubsub/listener.c
new file mode 100644
index 000000000..ed103872c
--- /dev/null
+++ b/src/storage/pubsub/listener.c
@@ -0,0 +1,218 @@
+#include "hashtable.h"
+#include "listener.h"
+
+#include <cc_mm.h>
+#include <cc_queue.h>
+#include <hash/cc_lookup3.h>
+
+static struct listener_slh *
+_ht_alloc(uint32_t nentry)
+{
+    struct listener_slh *table;
+    uint32_t i;
+
+    table = cc_alloc(sizeof(*table) * nentry);
+
+    if (table != NULL) {
+        for (i = 0; i < nentry; ++i) {
+            SLIST_INIT(&table[i]);
+        }
+    }
+
+    return table;
+}
+
+struct listener_ht *
+listener_ht_create(uint32_t hash_power)
+{
+    struct listener_ht *ht;
+    uint32_t nentry;
+
+    ASSERT(hash_power > 0);
+
+    ht = cc_alloc(sizeof(struct listener_ht));
+    if (ht == NULL) {
+        return NULL;
+    }
+
+    ht->hash_power = hash_power;
+    ht->nlistener = 0;
+    nentry = HASHSIZE(ht->hash_power);
+    ht->table = _ht_alloc(nentry);
+    if (ht->table == NULL) {
+        cc_free(ht);
+        return NULL;
+    }
+
+    return ht;
+}
+
+void listener_ht_destroy(struct listener_ht **ht)
+{
+    ASSERT(ht != NULL);
+
+    if (*ht != NULL && (*ht)->table != NULL) {
+        cc_free((*ht)->table);
+    }
+
+    cc_free(*ht);
+    *ht = NULL;
+}
+
+
+static struct listener_slh *
+_get_bucket(const channel_p ch, struct listener_ht *ht)
+{
+    /* use the _address_ of the channel to hash */
+    uint32_t hval = hash_lookup3((char *)&ch, sizeof(channel_p), 0);
+    return &(ht->table[hval & HASHMASK(ht->hash_power)]);
+}
+
+struct listener *
+listener_ht_get(const channel_p ch, struct listener_ht *ht)
+{
+    struct listener_slh *bucket;
+    struct listener *l;
+
+    bucket = _get_bucket(ch, ht);
+    for (l = SLIST_FIRST(bucket); l != NULL; l = SLIST_NEXT(l, l_sle)) {
+        if (l->ch == ch) {
+            return l;
+        }
+    }
+
+    return NULL;
+}
+
+void
+listener_ht_put(const struct listener *l, struct listener_ht *ht)
+{
+    struct listener_slh *bucket;
+
+    ASSERT(listener_ht_get(l->ch, ht) == NULL);
+
+    bucket = _get_bucket(l->ch, ht);
+    SLIST_INSERT_HEAD(bucket, (struct listener *)l, l_sle);
+
+    ht->nlistener++;
+}
+
+void
+listener_ht_delete(const channel_p ch, struct listener_ht *ht)
+{
+    struct listener_slh *bucket;
+    struct listener *l, *prev;
+
+    bucket = _get_bucket(ch, ht);
+    for (prev = NULL, l = SLIST_FIRST(bucket); l != NULL;
+        prev = l, l = SLIST_NEXT(l, l_sle)) {
+        if (l->ch == ch) {
+            break;
+        }
+    }
+
+    if (prev == NULL) {
+        SLIST_REMOVE_HEAD(bucket, l_sle);
+    } else {
+        SLIST_REMOVE_AFTER(prev, l_sle);
+    }
+
+    --(ht->nlistener);
+}
+
+
+struct listener *
+listener_create(channel_p ch, channel_handler_st *handler)
+{
+    struct listener *l;
+
+    l = cc_alloc(sizeof(struct listener));
+    if (l == NULL) {
+        return NULL;
+    }
+
+    l->idx = cc_alloc(sizeof(struct index_tqh));
+    if (l->idx == NULL) {
+        cc_free(l);
+        return NULL;
+    }
+
+    listener_reset(l);
+    l->ch = ch;
+    l->handler = handler;
+
+    return l;
+}
+
+void
+listener_destroy(struct listener **l)
+{
+    ASSERT(l != NULL && *l != NULL);
+
+    struct index_node *curr, *next;
+    struct index_tqh *idx = (*l)->idx;
+
+    /* delete all elements of the index */
+    TAILQ_FOREACH_SAFE(curr, idx, i_tqe, next) {
+        TAILQ_REMOVE(idx, curr, i_tqe);
+        cc_free(curr);
+    }
+    cc_free(idx);
+
+    cc_free(*l);
+    *l = NULL;
+}
+
+void
+listener_reset(struct listener *l)
+{
+    l->ch = NULL;
+    l->handler = NULL;
+    l->ntopic = 0;
+    TAILQ_INIT(l->idx);
+}
+
+bool
+listener_add_topic(struct listener *l, const struct topic *t)
+{
+    struct index_node *node;
+
+    ASSERT(l != NULL && t != NULL);
+
+    /* do nothing if already subscribed */
+    TAILQ_FOREACH(node, l->idx, i_tqe) {
+        if (node->obj == t) {
+            return false;
+        }
+    }
+
+    node = cc_alloc(sizeof(struct index_node));
+    if (node == NULL) {
+        return false;
+    }
+    node->obj = (struct topic *)t;
+
+    TAILQ_INSERT_TAIL(l->idx, node, i_tqe);
+    l->ntopic++;
+
+    return true;
+}
+
+void
+listener_del_topic(struct listener *l, const struct topic *t)
+{
+    struct index_node *node;
+
+    /* do nothing if not found */
+    TAILQ_FOREACH(node, l->idx, i_tqe) {
+        if (node->obj == t) {
+            break;
+        }
+    }
+    if (node == NULL) {
+        return;
+    }
+
+    TAILQ_REMOVE(l->idx, node, i_tqe);
+    l->ntopic--;
+}
diff --git a/src/storage/pubsub/listener.h b/src/storage/pubsub/listener.h
new file mode 100644
index 000000000..872ade1c2
--- /dev/null
+++ b/src/storage/pubsub/listener.h
@@ -0,0 +1,41 @@
+#pragma once
+
+#include "index.h"
+
+#include <channel/cc_channel.h>
+
+/*
+ * a listener is a client that has subscribed to at least one channel
+ */
+struct listener {
+    SLIST_ENTRY(listener)   l_sle;
+
+    channel_p               ch;
+    channel_handler_st      *handler;
+    uint32_t                ntopic;
+    struct index_tqh        *idx; /* index of all topics */
+};
+
+SLIST_HEAD(listener_slh, listener);
+
+struct listener_ht {
+    struct listener_slh *table;
+    uint32_t nlistener;
+    uint32_t hash_power;
+};
+
+struct topic;
+
+struct listener_ht *listener_ht_create(uint32_t hash_power);
+void listener_ht_destroy(struct listener_ht **ht);
+
+struct listener *listener_ht_get(const channel_p ch, struct listener_ht *ht);
+void listener_ht_put(const struct listener *l, struct listener_ht *ht);
+void listener_ht_delete(const channel_p ch, struct listener_ht *ht);
+
+struct listener *listener_create(channel_p ch, channel_handler_st *handler);
+void listener_destroy(struct listener **l);
+void listener_reset(struct listener *l);
+
+bool listener_add_topic(struct listener *l, const struct topic *t);
+void listener_del_topic(struct listener *l, const struct topic *t);
diff --git a/src/storage/pubsub/topic.c b/src/storage/pubsub/topic.c
new file mode 100644
index 000000000..77b40ff15
--- /dev/null
+++ b/src/storage/pubsub/topic.c
@@ -0,0 +1,218 @@
+#include "hashtable.h"
+#include "topic.h"
+
+#include <cc_mm.h>
+#include <cc_queue.h>
+#include <hash/cc_lookup3.h>
+
+static struct topic_slh *
+_ht_alloc(uint32_t nentry)
+{
+    struct topic_slh *table;
+    uint32_t i;
+
+    table = cc_alloc(sizeof(*table) * nentry);
+
+    if (table != NULL) {
+        for (i = 0; i < nentry; ++i) {
+            SLIST_INIT(&table[i]);
+        }
+    }
+
+    return table;
+}
+
+struct topic_ht *
+topic_ht_create(uint32_t hash_power)
+{
+    struct topic_ht *ht;
+    uint32_t nentry;
+
+    ASSERT(hash_power > 0);
+
+    ht = cc_alloc(sizeof(struct topic_ht));
+    if (ht == NULL) {
+        return NULL;
+    }
+
+    ht->hash_power = hash_power;
+    ht->ntopic = 0;
+    nentry = HASHSIZE(ht->hash_power);
+    ht->table = _ht_alloc(nentry);
+    if (ht->table == NULL) {
+        cc_free(ht);
+        return NULL;
+    }
+
+    return ht;
+}
+
+void
+topic_ht_destroy(struct topic_ht **ht)
+{
+    ASSERT(ht != NULL);
+
+    if (*ht != NULL && (*ht)->table != NULL) {
+        cc_free((*ht)->table);
+    }
+
+    cc_free(*ht);
+    *ht = NULL;
+}
+
+
+static struct topic_slh *
+_get_bucket(const struct bstring *name, struct topic_ht *ht)
+{
+    /* use the _address_ of the channel to hash */
+    uint32_t hval = hash_lookup3(name->data, name->len, 0);
+    return &(ht->table[hval & HASHMASK(ht->hash_power)]);
+}
+
+struct topic *
+topic_ht_get(const struct bstring *name, struct topic_ht *ht)
+{
+    struct topic_slh *bucket;
+    struct topic *t;
+
+    bucket = _get_bucket(name, ht);
+    for (t = SLIST_FIRST(bucket); t != NULL; t = SLIST_NEXT(t, t_sle)) {
+        if (bstring_compare(&t->name, name) == 0) {
+            return t;
+        }
+    }
+
+    return NULL;
+}
+
+void
+topic_ht_put(const struct topic *t, struct topic_ht *ht)
+{
+    struct topic_slh *bucket;
+
+    ASSERT(topic_ht_get(&t->name, ht) == NULL);
+
+    bucket = _get_bucket(&t->name, ht);
+    SLIST_INSERT_HEAD(bucket, (struct topic *)t, t_sle);
+
+    ht->ntopic++;
+}
+
+void
+topic_ht_delete(const struct bstring *name, struct topic_ht *ht)
+{
+    struct topic_slh *bucket;
+    struct topic *t, *prev;
+
+    bucket = _get_bucket(name, ht);
+    for (prev = NULL, t = SLIST_FIRST(bucket); t != NULL;
+        prev = t, t = SLIST_NEXT(t, t_sle)) {
+        if (bstring_compare(&t->name, name) == 0) {
+            break;
+        }
+    }
+
+    if (prev == NULL) {
+        SLIST_REMOVE_HEAD(bucket, t_sle);
+    } else {
+        SLIST_REMOVE_AFTER(prev, t_sle);
+    }
+
+    --(ht->ntopic);
+}
+
+
+struct topic *
+topic_create(const struct bstring *name)
+{
+    struct topic *t;
+
+    t = cc_alloc(sizeof(struct topic));
+    if (t == NULL) {
+        return NULL;
+    }
+
+    t->idx = cc_alloc(sizeof(struct index_tqh));
+    if (t->idx == NULL) {
+        cc_free(t);
+        return NULL;
+    }
+
+    topic_reset(t);
+    t->name = *name;
+
+    return t;
+}
+
+void
+topic_destroy(struct topic **t)
+{
+    ASSERT(t != NULL && *t != NULL);
+
+    struct index_node *curr, *next;
+    struct index_tqh *idx = (*t)->idx;
+
+    /* delete all elements of the index */
+    TAILQ_FOREACH_SAFE(curr, idx, i_tqe, next) {
+        TAILQ_REMOVE(idx, curr, i_tqe);
+        cc_free(curr);
+    }
+    cc_free(idx);
+
+    cc_free(*t);
+    *t = NULL;
+}
+
+void
+topic_reset(struct topic *t)
+{
+    t->name = null_bstring;
+    t->nsub = 0;
+    TAILQ_INIT(t->idx);
+}
+
+bool
+topic_add_listener(struct topic *t, const struct listener *l)
+{
+    struct index_node *node;
+
+    ASSERT(t != NULL && l != NULL);
+
+    /* do nothing if already subscribed */
+    TAILQ_FOREACH(node, t->idx, i_tqe) {
+        if (node->obj == l) {
+            return false;
+        }
+    }
+
+    node = cc_alloc(sizeof(struct index_node));
+    if (node == NULL) {
+        return false;
+    }
+    node->obj = (struct listener *)l;
+
+    TAILQ_INSERT_TAIL(t->idx, node, i_tqe);
+    t->nsub++;
+
+    return true;
+}
+
+void
+topic_del_listener(struct topic *t, const struct listener *l)
+{
+    struct index_node *node;
+
+    /* do nothing if not found */
+    TAILQ_FOREACH(node, t->idx, i_tqe) {
+        if (node->obj == l) {
+            break;
+        }
+    }
+    if (node == NULL) {
+        return;
+    }
+
+    TAILQ_REMOVE(t->idx, node, i_tqe);
+    t->nsub--;
+    cc_free(node);
+}
diff --git a/src/storage/pubsub/topic.h b/src/storage/pubsub/topic.h
new file mode 100644
index 000000000..f9ad7d9f2
--- /dev/null
+++ b/src/storage/pubsub/topic.h
@@ -0,0 +1,41 @@
+#pragma once
+
+#include "index.h"
+
+#include <cc_bstring.h>
+
+/*
+ * a topic is an endpoint that clients can subscribe to, equivalent to
+ * "channel" in the original redis protocol.
+ */
+struct topic {
+    SLIST_ENTRY(topic)  t_sle;
+
+    struct bstring      name;
+    uint32_t            nsub;
+    struct index_tqh    *idx; /* index of all listeners */
+};
+
+SLIST_HEAD(topic_slh, topic);
+
+struct topic_ht {
+    struct topic_slh *table;
+    uint32_t ntopic;
+    uint32_t hash_power;
+};
+
+struct listener;
+
+struct topic_ht *topic_ht_create(uint32_t hash_power);
+void topic_ht_destroy(struct topic_ht **ht);
+
+struct topic *topic_ht_get(const struct bstring *name, struct topic_ht *ht);
+void topic_ht_put(const struct topic *t, struct topic_ht *ht);
+void topic_ht_delete(const struct bstring *name, struct topic_ht *ht);
+
+struct topic *topic_create(const struct bstring *name);
+void topic_destroy(struct topic **t);
+void topic_reset(struct topic *t);
+
+bool topic_add_listener(struct topic *t, const struct listener *l);
+void topic_del_listener(struct topic *t, const struct listener *l);

From 8fce409978860b970e41ba411ca13401d1cca454 Mon Sep 17 00:00:00 2001
From: Yao Yue <yao@twitter.com>
Date: Thu, 6 Apr 2017 12:52:52 -0700
Subject: [PATCH 2/9] add pubsub server

---
 CMakeLists.txt                         |   7 +-
 config/pubsub.conf                     |   7 +
 src/core/data/worker.c                 |   2 +-
 src/protocol/data/redis/request.c      |   1 +
 src/server/CMakeLists.txt              |   4 +
 src/server/pubsub/CMakeLists.txt       |  29 ++++
 src/server/pubsub/admin/CMakeLists.txt |   4 +
 src/server/pubsub/admin/process.c      |  68 +++++++++
 src/server/pubsub/admin/process.h      |   4 +
 src/server/pubsub/data/CMakeLists.txt  |   5 +
 src/server/pubsub/data/process.c       | 150 +++++++++++++++++++
 src/server/pubsub/data/process.h       |  30 ++++
 src/server/pubsub/data/pubsub.c        |  59 ++++++++
 src/server/pubsub/data/pubsub.h        |  12 ++
 src/server/pubsub/main.c               | 197 +++++++++++++++++++++++++
 src/server/pubsub/setting.c            |  18 +++
 src/server/pubsub/setting.h            |  50 +++++++
 src/server/pubsub/stats.c              |  21 +++
 src/server/pubsub/stats.h              |  38 +++++
 19 files changed, 702 insertions(+), 4 deletions(-)
 create mode 100644 config/pubsub.conf
 create mode 100644 src/server/pubsub/CMakeLists.txt
 create mode 100644 src/server/pubsub/admin/CMakeLists.txt
 create mode 100644 src/server/pubsub/admin/process.c
 create mode 100644 src/server/pubsub/admin/process.h
 create mode 100644 src/server/pubsub/data/CMakeLists.txt
 create mode 100644 src/server/pubsub/data/process.c
 create mode 100644 src/server/pubsub/data/process.h
 create mode 100644 src/server/pubsub/data/pubsub.c
 create mode 100644 src/server/pubsub/data/pubsub.h
 create mode 100644 src/server/pubsub/main.c
 create mode 100644 src/server/pubsub/setting.c
 create mode 100644 src/server/pubsub/setting.h
 create mode 100644 src/server/pubsub/stats.c
 create mode 100644 src/server/pubsub/stats.h

diff --git a/CMakeLists.txt b/CMakeLists.txt
index 10dc97edc..888128019 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -33,9 +33,10 @@ option(HAVE_ASSERT_LOG "assert_log enabled by default" ON)
 option(HAVE_ASSERT_PANIC "assert_panic disabled by default" OFF)
 option(HAVE_LOGGING "logging enabled by default" ON)
 option(HAVE_STATS "stats enabled by default" ON)
-option(TARGET_PINGSERVER "build pingserver binary" ON)
-option(TARGET_SLIMCACHE "build slimcache binary" ON)
-option(TARGET_TWEMCACHE "build twemcache binary" ON)
+option(TARGET_PINGSERVER "build pingserver binary" OFF)
+option(TARGET_PUBSUB "build pubsub binary" ON)
+option(TARGET_SLIMCACHE "build slimcache binary" OFF)
+option(TARGET_TWEMCACHE "build twemcache binary" OFF)
 option(COVERAGE "code coverage" OFF)
 
 # Note: duplicate custom targets only works with Makefile generators, will break XCode & VS
diff --git a/config/pubsub.conf b/config/pubsub.conf
new file mode 100644
index 000000000..90d333165
--- /dev/null
+++ b/config/pubsub.conf
@@ -0,0 +1,7 @@
+# if slab profile is specified, then the profile wil be explicitly set
+# otherwise, slab profile will automatically be generated by using growth factor, slab size, and chunk size
+#slab_profile: 1024 2048 4096 8192 16384 32768 65536 131072 262144 524288
+
+debug_log_level: 6
+# debug_log_file: pubsub.log
+# debug_log_nbuf: 16384
diff --git a/src/core/data/worker.c b/src/core/data/worker.c
index 473922c31..336fbd058 100644
--- a/src/core/data/worker.c
+++ b/src/core/data/worker.c
@@ -291,7 +291,7 @@ _worker_evwait(void)
 void *
 core_worker_evloop(void *arg)
 {
-    processor = arg;
+    post_processor = arg;
 
     for(;;) {
         if (_worker_evwait() != CC_OK) {
diff --git a/src/protocol/data/redis/request.c b/src/protocol/data/redis/request.c
index 2a38e5c46..2eec939df 100644
--- a/src/protocol/data/redis/request.c
+++ b/src/protocol/data/redis/request.c
@@ -19,6 +19,7 @@ struct command command_table[REQ_SENTINEL] = {
     REQ_HASH(CMD_INIT)
     REQ_ZSET(CMD_INIT)
     REQ_MISC(CMD_INIT)
+    REQ_PUBSUB(CMD_INIT)
 };
 #undef CMD_INIT
 
diff --git a/src/server/CMakeLists.txt b/src/server/CMakeLists.txt
index 9dff0246e..f6043c56e 100644
--- a/src/server/CMakeLists.txt
+++ b/src/server/CMakeLists.txt
@@ -2,6 +2,10 @@ if(TARGET_PINGSERVER)
     add_subdirectory(pingserver)
 endif()
 
+if(TARGET_PUBSUB)
+    add_subdirectory(pubsub)
+endif()
+
 if(TARGET_SLIMCACHE)
     add_subdirectory(slimcache)
 endif()
diff --git a/src/server/pubsub/CMakeLists.txt b/src/server/pubsub/CMakeLists.txt
new file mode 100644
index 000000000..f30ed9f30
--- /dev/null
+++ b/src/server/pubsub/CMakeLists.txt
@@ -0,0 +1,29 @@
+add_subdirectory(admin)
+add_subdirectory(data)
+
+set(SOURCE
+    ${SOURCE}
+    main.c
+    setting.c
+    stats.c)
+
+set(MODULES
+    core
+    protocol_admin
+    protocol_redis
+    pubsub
+    time
+    util)
+
+set(LIBS
+    ccommon-static
+    ${CMAKE_THREAD_LIBS_INIT})
+
+set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${PROJECT_BINARY_DIR}/_bin)
+set(TARGET_NAME ${PROJECT_NAME}_pubsub)
+
+add_executable(${TARGET_NAME} ${SOURCE})
+target_link_libraries(${TARGET_NAME} ${MODULES} ${LIBS})
+
+install(TARGETS ${TARGET_NAME} RUNTIME DESTINATION bin)
+add_dependencies(service ${TARGET_NAME})
diff --git a/src/server/pubsub/admin/CMakeLists.txt b/src/server/pubsub/admin/CMakeLists.txt
new file mode 100644
index 000000000..31a7e65f3
--- /dev/null
+++ b/src/server/pubsub/admin/CMakeLists.txt
@@ -0,0 +1,4 @@
+set(SOURCE
+    ${SOURCE}
+    ${CMAKE_CURRENT_SOURCE_DIR}/process.c
+    PARENT_SCOPE)
diff --git a/src/server/pubsub/admin/process.c b/src/server/pubsub/admin/process.c
new file mode 100644
index 000000000..00f36d060
--- /dev/null
+++ b/src/server/pubsub/admin/process.c
@@ -0,0 +1,68 @@
+#include "process.h"
+
+#include "protocol/admin/admin_include.h"
+#include "util/procinfo.h"
+
+#include <cc_mm.h>
+
+#define PUBSUB_ADMIN_MODULE_NAME "pubsub::admin"
+
+extern struct stats stats;
+extern unsigned int nmetric;
+
+static bool admin_init = false;
+static char *buf = NULL;
+static size_t cap;
+
+void
+admin_process_setup(void)
+{
+    log_info("set up the %s module", PUBSUB_ADMIN_MODULE_NAME);
+    if (admin_init) {
+        log_warn("%s has already been setup, overwrite",
+                 PUBSUB_ADMIN_MODULE_NAME);
+    }
+
+    cap = METRIC_PRINT_LEN * nmetric + METRIC_END_LEN;
+    buf = cc_alloc(cap);
+    /* TODO: check return status of cc_alloc */
+
+    admin_init = true;
+}
+
+void
+admin_process_teardown(void)
+{
+    log_info("tear down the %s module", PUBSUB_ADMIN_MODULE_NAME);
+    if (!admin_init) {
+        log_warn("%s has never been setup", PUBSUB_ADMIN_MODULE_NAME);
+    }
+
+    admin_init = false;
+}
+
+static void
+_admin_stats(struct response *rsp, struct request *req)
+{
+    procinfo_update();
+    rsp->data.data = buf;
+    rsp->data.len = print_stats(buf, cap, (struct metric *)&stats, nmetric);
+}
+
+void
+admin_process_request(struct response *rsp, struct request *req)
+{
+    rsp->type = RSP_GENERIC;
+
+    switch (req->type) {
+    case REQ_STATS:
+        _admin_stats(rsp, req);
+        break;
+    case REQ_VERSION:
+        rsp->data = str2bstr(VERSION_PRINTED);
+        break;
+    default:
+        rsp->type = RSP_INVALID;
+        break;
+    }
+}
diff --git a/src/server/pubsub/admin/process.h b/src/server/pubsub/admin/process.h
new file mode 100644
index 000000000..361230004
--- /dev/null
+++ b/src/server/pubsub/admin/process.h
@@ -0,0 +1,4 @@
+#pragma once
+
+void admin_process_setup(void);
+void admin_process_teardown(void);
diff --git a/src/server/pubsub/data/CMakeLists.txt b/src/server/pubsub/data/CMakeLists.txt
new file mode 100644
index 000000000..ff285f978
--- /dev/null
+++ b/src/server/pubsub/data/CMakeLists.txt
@@ -0,0 +1,5 @@
+set(SOURCE
+    ${SOURCE}
+    ${CMAKE_CURRENT_SOURCE_DIR}/process.c
+    ${CMAKE_CURRENT_SOURCE_DIR}/pubsub.c
+    PARENT_SCOPE)
diff --git a/src/server/pubsub/data/process.c b/src/server/pubsub/data/process.c
new file mode 100644
index 000000000..2ac1a8ad6
--- /dev/null
+++ b/src/server/pubsub/data/process.c
@@ -0,0 +1,150 @@
+#include "process.h"
+
+#include "protocol/data/redis_include.h"
+#include "pubsub.h"
+
+#include <buffer/cc_dbuf.h>
+#include <cc_debug.h>
+
+#define PUBSUB_PROCESS_MODULE_NAME "pubsub::process"
+
+command_fn command_registry[REQ_SENTINEL];
+
+static bool process_init = false;
+static process_metrics_st *process_metrics = NULL;
+
+
+void
+process_setup(process_metrics_st *metrics)
+{
+    log_info("set up the %s module", PUBSUB_PROCESS_MODULE_NAME);
+
+    if (process_init) {
+        log_warn("%s has already been setup, overwrite",
+                 PUBSUB_PROCESS_MODULE_NAME);
+    }
+
+    pubsub_setup();
+
+    command_registry[REQ_SUBSCRIBE] = command_subscribe;
+
+    process_metrics = metrics;
+    process_init = true;
+}
+
+void
+process_teardown(void)
+{
+    log_info("tear down the %s module", PUBSUB_PROCESS_MODULE_NAME);
+    if (!process_init) {
+        log_warn("%s has never been setup", PUBSUB_PROCESS_MODULE_NAME);
+    }
+
+    pubsub_teardown();
+
+    command_registry[REQ_SUBSCRIBE] = NULL;
+
+    process_metrics = NULL;
+    process_init = false;
+}
+
+static void
+process_request_sock(struct response *rsp, struct request *req, struct buf_sock *s)
+{
+    log_verb("processing req %p, write rsp to %p", req, rsp);
+    INCR(process_metrics, process_req);
+
+    if (command_registry[req->type] == NULL) {
+        /* return error */
+    }
+
+    command_registry[req->type](rsp, req, s);
+}
+
+int
+pubsub_process_read(struct buf_sock *s)
+{
+    int status;
+    struct request *req;
+    struct response *rsp;
+
+    log_verb("post-read processing");
+
+    req = request_borrow();
+    rsp = response_borrow();
+
+    /* keep parse-process-compose until running out of data in rbuf */
+    while (buf_rsize(s->rbuf) > 0) {
+        /* stage 1: parsing */
+        log_verb("%"PRIu32" bytes left", buf_rsize(s->rbuf));
+
+        status = parse_req(req, s->rbuf);
+        if (status == PARSE_EUNFIN) {
+            buf_lshift(s->rbuf);
+            return 0;
+        }
+        if (status != PARSE_OK) {
+            /* parsing errors are all client errors, since we don't know
+             * how to recover from client errors in this condition (we do not
+             * have a valid request so we don't know where the invalid request
+             * ends), we should close the connection
+             */
+            log_warn("illegal request received, status: %d", status);
+            return -1;
+        }
+
+        /* stage 2: processing- check for quit, allocate response(s), process */
+
+        /* quit is special, no processing/resposne expected */
+        if (req->type == REQ_QUIT) {
+            log_info("peer called quit");
+            return -1;
+        }
+
+        /* actual processing */
+        process_request_sock(rsp, req, s);
+
+        /* stage 3: write response(s) if necessary */
+
+        /* noreply means no need to write to buffers */
+
+        /* logging, clean-up */
+    }
+
+    request_return(&req);
+    response_return(&rsp);
+
+    return 0;
+}
+
+
+int
+pubsub_process_write(struct buf_sock *s)
+{
+    log_verb("post-write processing");
+
+    buf_lshift(s->rbuf);
+    buf_lshift(s->wbuf);
+    dbuf_shrink(&s->rbuf);
+    dbuf_shrink(&s->wbuf);
+
+    return 0;
+}
+
+
+int
+pubsub_process_error(struct buf_sock *s)
+{
+    struct request *req;
+    struct response *rsp;
+
+    log_verb("post-error processing");
+
+    /* normalize buffer size */
+    buf_reset(s->rbuf);
+    dbuf_shrink(&s->rbuf);
+    buf_reset(s->wbuf);
+    dbuf_shrink(&s->wbuf);
+
+    return 0;
+}
diff --git a/src/server/pubsub/data/process.h b/src/server/pubsub/data/process.h
new file mode 100644
index 000000000..63ec79ddc
--- /dev/null
+++ b/src/server/pubsub/data/process.h
@@ -0,0 +1,30 @@
+#pragma once
+
+#include "protocol/data/redis_include.h"
+
+#include <buffer/cc_buf.h>
+#include <cc_metric.h>
+#include <cc_option.h>
+#include <stream/cc_sockio.h>
+
+/*          name                        type            description */
+#define PROCESS_METRIC(ACTION)                                          \
+    ACTION( process_req,       METRIC_COUNTER, "# requests processed"  )\
+    ACTION( process_ex,        METRIC_COUNTER, "# processing error"    )\
+    ACTION( publish,           METRIC_COUNTER, "# publish requests"    )\
+    ACTION( subscribe,         METRIC_COUNTER, "# subscribe requests"  )\
+    ACTION( unsubscribe,       METRIC_COUNTER, "# unsubscribe requests")
+
+typedef struct {
+    PROCESS_METRIC(METRIC_DECLARE)
+} process_metrics_st;
+
+typedef void (* command_fn)(struct response *, struct request *, struct buf_sock *);
+extern command_fn command_registry[REQ_SENTINEL];
+
+void process_setup(process_metrics_st *metrics);
+void process_teardown(void);
+
+int pubsub_process_read(struct buf_sock *s);
+int pubsub_process_write(struct buf_sock *s);
+int pubsub_process_error(struct buf_sock *s);
diff --git a/src/server/pubsub/data/pubsub.c b/src/server/pubsub/data/pubsub.c
new file mode 100644
index 000000000..cd35ed595
--- /dev/null
+++ b/src/server/pubsub/data/pubsub.c
@@ -0,0 +1,59 @@
+#include "pubsub.h"
+
+#include "protocol/data/redis_include.h"
+#include "storage/pubsub/listener.h"
+#include "storage/pubsub/topic.h"
+
+#include <cc_array.h>
+#include <stream/cc_sockio.h>
+
+static struct listener_ht *lht;
+static struct topic_ht *tht;
+
+void
+pubsub_setup(void)
+{
+    lht = listener_ht_create(16);
+    tht = topic_ht_create(16);
+}
+
+void
+pubsub_teardown(void)
+{
+    listener_ht_destroy(&lht);
+    topic_ht_destroy(&tht);
+}
+
+/* "subscribe topic [topic ...]" */
+void
+command_subscribe(struct response *rsp, struct request *req, struct buf_sock *s)
+{
+    struct element *el;
+    struct listener *l;
+    struct topic *t;
+    uint32_t ntopic = req->token->nelem - 1;
+
+    l = listener_ht_get(s->ch, lht);
+    if (l == NULL) {
+        l = listener_create(s->ch, s->hdl);
+        listener_ht_put(l, lht);
+    }
+
+    for (int i = 1; i < ntopic; i++) {
+        el = array_get(req->token, i);
+        if (el->type != ELEM_BULK) {
+            /* handle client error */
+        };
+
+        t = topic_ht_get(&el->bstr, tht);
+        if (t == NULL) {
+            t = topic_create(&el->bstr);
+        }
+        listener_add_topic(l, t);
+    }
+
+    rsp->type = ELEM_STR;
+    el = array_push(rsp->token);
+    el->type = ELEM_STR;
+    el->bstr = str2bstr(RSP_STR_OK);
+}
diff --git a/src/server/pubsub/data/pubsub.h b/src/server/pubsub/data/pubsub.h
new file mode 100644
index 000000000..72df15661
--- /dev/null
+++ b/src/server/pubsub/data/pubsub.h
@@ -0,0 +1,12 @@
+#pragma once
+
+struct request;
+struct response;
+struct buf_sock;
+
+void pubsub_setup(void);
+void pubsub_teardown(void);
+
+void command_subscribe(struct response *rsp, struct request *req, struct buf_sock *s);
+void command_unsubscribe(struct response *rsp, struct request *req, struct buf_sock *s);
+void command_publish(struct response *rsp, struct request *req, struct buf_sock *s);
diff --git a/src/server/pubsub/main.c b/src/server/pubsub/main.c
new file mode 100644
index 000000000..411f5fc9b
--- /dev/null
+++ b/src/server/pubsub/main.c
@@ -0,0 +1,197 @@
+#include "setting.h"
+#include "stats.h"
+
+#include "time/time.h"
+#include "util/util.h"
+
+#include <cc_debug.h>
+
+#include <errno.h>
+#include <fcntl.h>
+#include <stdio.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <sysexits.h>
+
+struct processor worker_processor = {
+    pubsub_process_read,
+    pubsub_process_write,
+    pubsub_process_error,
+};
+
+static void
+show_usage(void)
+{
+    log_stdout(
+            "Usage:" CRLF
+            "  pelikan_pubsub [option|config]" CRLF
+            );
+    log_stdout(
+            "Description:" CRLF
+            "  pelikan_pubsub is an in-memory pub/sub server." CRLF
+            CRLF
+            "  It supports basic Redis pub/sub commands:" CRLF
+            "    subscribe, unsubscribe, publish,... " CRLF
+            );
+    log_stdout(
+            "Command-line options:" CRLF
+            "  -h, --help        show this message" CRLF
+            "  -v, --version     show version number" CRLF
+            "  -c, --config      list & describe all options in config" CRLF
+            "  -s, --stats       list & describe all metrics in stats" CRLF
+            );
+    log_stdout(
+            "Example:" CRLF
+            "  pelikan_pubsub pubsub.conf" CRLF CRLF
+            "Sample config files can be found under the config dir." CRLF
+            );
+}
+
+static void
+teardown(void)
+{
+    core_teardown();
+    admin_process_teardown();
+    compose_teardown();
+    parse_teardown();
+    procinfo_teardown();
+    time_teardown();
+
+    timing_wheel_teardown();
+    tcp_teardown();
+    sockio_teardown();
+    event_teardown();
+    dbuf_teardown();
+    buf_teardown();
+
+    debug_teardown();
+    log_teardown();
+}
+
+static void
+setup(void)
+{
+    char *fname = NULL;
+    uint64_t intvl;
+
+    if (atexit(teardown) != 0) {
+        log_stderr("cannot register teardown procedure with atexit()");
+        exit(EX_OSERR); /* only failure comes from NOMEM */
+    }
+
+    /* Setup logging first */
+    log_setup(&stats.log);
+    if (debug_setup(&setting.debug) < 0) {
+        log_stderr("debug log setup failed");
+        goto error;
+    }
+
+    /* setup top-level application options */
+    if (option_bool(&setting.pubsub.daemonize)) {
+        daemonize();
+    }
+    fname = option_str(&setting.pubsub.pid_filename);
+    if (fname != NULL) {
+        /* to get the correct pid, call create_pidfile after daemonize */
+        create_pidfile(fname);
+    }
+
+    /* setup library modules */
+    buf_setup(&setting.buf, &stats.buf);
+    dbuf_setup(&setting.dbuf, &stats.dbuf);
+    event_setup(&stats.event);
+    sockio_setup(&setting.sockio, &stats.sockio);
+    tcp_setup(&setting.tcp, &stats.tcp);
+    timing_wheel_setup(&stats.timing_wheel);
+
+    /* setup pelikan modules */
+    time_setup();
+    procinfo_setup(&stats.procinfo);
+    request_setup(&setting.request, &stats.request);
+    response_setup(&setting.response, &stats.response);
+    parse_setup(&stats.parse_req, NULL);
+    compose_setup(NULL, &stats.compose_rsp);
+    process_setup(&stats.process);
+    admin_process_setup();
+    core_setup(&setting.admin, &setting.server, &setting.worker,
+            &stats.server, &stats.worker);
+
+    /* adding recurring events to maintenance/admin thread */
+    intvl = option_uint(&setting.pubsub.dlog_intvl);
+    if (core_admin_register(intvl, debug_log_flush, NULL) == NULL) {
+        log_stderr("Could not register timed event to flush debug log");
+        goto error;
+    }
+
+    return;
+
+error:
+    if (fname != NULL) {
+        remove_pidfile(fname);
+    }
+
+    /* since we registered teardown with atexit, it'll be called upon exit */
+    exit(EX_CONFIG);
+}
+
+int
+main(int argc, char **argv)
+{
+    rstatus_i status = CC_OK;
+    FILE *fp = NULL;
+
+    if (argc > 2) {
+        show_usage();
+        exit(EX_USAGE);
+    }
+
+    if (argc == 1) {
+        log_stderr("launching server with default values.");
+    }
+
+    if (argc == 2) {
+        if (strcmp(argv[1], "-h") == 0 || strcmp(argv[1], "--help") == 0) {
+            show_usage();
+            exit(EX_OK);
+        }
+        if (strcmp(argv[1], "-v") == 0 || strcmp(argv[1], "--version") == 0) {
+            show_version();
+            exit(EX_OK);
+        }
+        if (strcmp(argv[1], "-c") == 0 || strcmp(argv[1], "--config") == 0) {
+            option_describe_all((struct option *)&setting, nopt);
+            exit(EX_OK);
+        }
+        if (strcmp(argv[1], "-s") == 0 || strcmp(argv[1], "--stats") == 0) {
+            metric_describe_all((struct metric *)&stats, nmetric);
+            exit(EX_OK);
+        }
+        fp = fopen(argv[1], "r");
+        if (fp == NULL) {
+            log_stderr("cannot open config: incorrect path or doesn't exist");
+            exit(EX_DATAERR);
+        }
+    }
+
+    if (option_load_default((struct option *)&setting, nopt) != CC_OK) {
+        log_stderr("failed to load default option values");
+        exit(EX_CONFIG);
+    }
+
+    if (fp != NULL) {
+        log_stderr("load config from %s", argv[1]);
+        status = option_load_file(fp, (struct option *)&setting, nopt);
+        fclose(fp);
+    }
+    if (status != CC_OK) {
+        log_stderr("failed to load config");
+        exit(EX_DATAERR);
+    }
+
+    setup();
+    option_print_all((struct option *)&setting, nopt);
+
+    core_run(NULL, &worker_processor);
+
+    exit(EX_OK);
+}
diff --git a/src/server/pubsub/setting.c b/src/server/pubsub/setting.c
new file mode 100644
index 000000000..6854a0e53
--- /dev/null
+++ b/src/server/pubsub/setting.c
@@ -0,0 +1,18 @@
+#include "setting.h"
+
+struct setting setting = {
+    { PUBSUB_OPTION(OPTION_INIT)    },
+    { ADMIN_OPTION(OPTION_INIT)     },
+    { SERVER_OPTION(OPTION_INIT)    },
+    { WORKER_OPTION(OPTION_INIT)    },
+    { REQUEST_OPTION(OPTION_INIT)   },
+    { RESPONSE_OPTION(OPTION_INIT)  },
+    { ARRAY_OPTION(OPTION_INIT)     },
+    { BUF_OPTION(OPTION_INIT)       },
+    { DBUF_OPTION(OPTION_INIT)      },
+    { DEBUG_OPTION(OPTION_INIT)     },
+    { SOCKIO_OPTION(OPTION_INIT)    },
+    { TCP_OPTION(OPTION_INIT)       },
+};
+
+unsigned int nopt = OPTION_CARDINALITY(setting);
diff --git a/src/server/pubsub/setting.h b/src/server/pubsub/setting.h
new file mode 100644
index 000000000..bb60374b6
--- /dev/null
+++ b/src/server/pubsub/setting.h
@@ -0,0 +1,50 @@
+#pragma once
+
+#include "admin/process.h"
+#include "data/process.h"
+
+#include "core/core.h"
+#include "protocol/data/redis_include.h"
+
+#include <buffer/cc_buf.h>
+#include <buffer/cc_dbuf.h>
+#include <cc_array.h>
+#include <cc_debug.h>
+#include <cc_metric.h>
+#include <cc_option.h>
+#include <cc_ring_array.h>
+#include <cc_signal.h>
+#include <channel/cc_tcp.h>
+#include <stream/cc_sockio.h>
+
+/* option related */
+/*          name            type                default description */
+#define PUBSUB_OPTION(ACTION)                                                        \
+    ACTION( daemonize,      OPTION_TYPE_BOOL,   false,  "daemonize the process"        )\
+    ACTION( pid_filename,   OPTION_TYPE_STR,    NULL,   "file storing the pid"         )\
+    ACTION( dlog_intvl,     OPTION_TYPE_UINT,   500,    "debug log flush interval(ms)" )
+
+typedef struct {
+    PUBSUB_OPTION(OPTION_DECLARE)
+} pubsub_options_st;
+
+struct setting {
+    /* top-level */
+    pubsub_options_st       pubsub;
+    /* application modules */
+    admin_options_st        admin;
+    server_options_st       server;
+    worker_options_st       worker;
+    request_options_st      request;
+    response_options_st     response;
+    /* ccommon libraries */
+    array_options_st        array;
+    buf_options_st          buf;
+    dbuf_options_st         dbuf;
+    debug_options_st        debug;
+    sockio_options_st       sockio;
+    tcp_options_st          tcp;
+};
+
+extern struct setting setting;
+extern unsigned int nopt;
diff --git a/src/server/pubsub/stats.c b/src/server/pubsub/stats.c
new file mode 100644
index 000000000..1c86a710a
--- /dev/null
+++ b/src/server/pubsub/stats.c
@@ -0,0 +1,21 @@
+#include "stats.h"
+
+struct stats stats = {
+    { PROCINFO_METRIC(METRIC_INIT)      },
+    { PROCESS_METRIC(METRIC_INIT)       },
+    { PARSE_REQ_METRIC(METRIC_INIT)     },
+    { COMPOSE_RSP_METRIC(METRIC_INIT)   },
+    { REQUEST_METRIC(METRIC_INIT)       },
+    { RESPONSE_METRIC(METRIC_INIT)      },
+    { CORE_SERVER_METRIC(METRIC_INIT)   },
+    { CORE_WORKER_METRIC(METRIC_INIT)   },
+    { BUF_METRIC(METRIC_INIT)           },
+    { DBUF_METRIC(METRIC_INIT)          },
+    { EVENT_METRIC(METRIC_INIT)         },
+    { LOG_METRIC(METRIC_INIT)           },
+    { SOCKIO_METRIC(METRIC_INIT)        },
+    { TCP_METRIC(METRIC_INIT)           },
+    { TIMING_WHEEL_METRIC(METRIC_INIT)  },
+};
+
+unsigned int nmetric = METRIC_CARDINALITY(stats);
diff --git a/src/server/pubsub/stats.h b/src/server/pubsub/stats.h
new file mode 100644
index 000000000..5c458fd4c
--- /dev/null
+++ b/src/server/pubsub/stats.h
@@ -0,0 +1,38 @@
+#pragma once
+
+#include "data/process.h"
+
+#include "core/core.h"
+#include "protocol/data/redis_include.h"
+#include "util/procinfo.h"
+
+#include <buffer/cc_dbuf.h>
+#include <cc_event.h>
+#include <cc_log.h>
+#include <channel/cc_tcp.h>
+#include <stream/cc_sockio.h>
+#include <time/cc_wheel.h>
+
+struct stats {
+    /* perf info */
+    procinfo_metrics_st         procinfo;
+    /* application modules */
+    process_metrics_st          process;
+    parse_req_metrics_st        parse_req;
+    compose_rsp_metrics_st      compose_rsp;
+    request_metrics_st          request;
+    response_metrics_st         response;
+    server_metrics_st           server;
+    worker_metrics_st           worker;
+    /* ccommon libraries */
+    buf_metrics_st              buf;
+    dbuf_metrics_st             dbuf;
+    event_metrics_st            event;
+    log_metrics_st              log;
+    sockio_metrics_st           sockio;
+    tcp_metrics_st              tcp;
+    timing_wheel_metrics_st     timing_wheel;
+};
+
+extern struct stats stats;
+extern unsigned int nmetric;

From 9be85d475a4591abc22b81c9f168eb71493c494f Mon Sep 17 00:00:00 2001
From: Yao Yue <yao@twitter.com>
Date: Sun, 9 Apr 2017 18:13:23 -0700
Subject: [PATCH 3/9] add response processing

---
 src/server/pubsub/data/process.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/src/server/pubsub/data/process.c b/src/server/pubsub/data/process.c
index 2ac1a8ad6..9ee2d5e91 100644
--- a/src/server/pubsub/data/process.c
+++ b/src/server/pubsub/data/process.c
@@ -105,6 +105,7 @@ pubsub_process_read(struct buf_sock *s)
         process_request_sock(rsp, req, s);
 
         /* stage 3: write response(s) if necessary */
+        compose_rsp(&s->wbuf, rsp);
 
         /* noreply means no need to write to buffers */
 

From 869d18ca0263a3ca0dc4136382b08c4355bd91e3 Mon Sep 17 00:00:00 2001
From: Yao Yue <yao@twitter.com>
Date: Mon, 3 Jul 2017 17:08:50 -0700
Subject: [PATCH 4/9] everything compiles

---
 CMakeLists.txt                        |  6 +--
 src/core/context.h                    |  1 +
 src/core/core.c                       | 31 +++++++++++++-
 src/core/core.h                       |  4 +-
 src/core/data/CMakeLists.txt          |  1 +
 src/core/data/worker.c                |  2 +-
 src/server/pingserver/main.c          |  2 +-
 src/server/pubsub/data/CMakeLists.txt |  1 -
 src/server/pubsub/data/process.c      | 49 ++++++++++++++++++++--
 src/server/pubsub/data/process.h      |  2 +
 src/server/pubsub/data/pubsub.c       | 59 ---------------------------
 src/server/pubsub/data/pubsub.h       | 12 ------
 src/server/pubsub/main.c              | 19 +++++----
 src/server/pubsub/setting.c           |  4 +-
 src/server/pubsub/setting.h           | 10 ++---
 src/server/pubsub/stats.c             |  2 +-
 src/server/pubsub/stats.h             |  2 +-
 src/server/slimcache/main.c           |  2 +-
 src/server/twemcache/main.c           |  2 +-
 19 files changed, 109 insertions(+), 102 deletions(-)
 delete mode 100644 src/server/pubsub/data/pubsub.c
 delete mode 100644 src/server/pubsub/data/pubsub.h

diff --git a/CMakeLists.txt b/CMakeLists.txt
index 888128019..4382ad713 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -33,10 +33,10 @@ option(HAVE_ASSERT_LOG "assert_log enabled by default" ON)
 option(HAVE_ASSERT_PANIC "assert_panic disabled by default" OFF)
 option(HAVE_LOGGING "logging enabled by default" ON)
 option(HAVE_STATS "stats enabled by default" ON)
-option(TARGET_PINGSERVER "build pingserver binary" OFF)
+option(TARGET_PINGSERVER "build pingserver binary" ON)
 option(TARGET_PUBSUB "build pubsub binary" ON)
-option(TARGET_SLIMCACHE "build slimcache binary" OFF)
-option(TARGET_TWEMCACHE "build twemcache binary" OFF)
+option(TARGET_SLIMCACHE "build slimcache binary" ON)
+option(TARGET_TWEMCACHE "build twemcache binary" ON)
 option(COVERAGE "code coverage" OFF)
 
 # Note: duplicate custom targets only works with Makefile generators, will break XCode & VS
diff --git a/src/core/context.h b/src/core/context.h
index f3d215eb4..4d705bec4 100644
--- a/src/core/context.h
+++ b/src/core/context.h
@@ -12,5 +12,6 @@ struct context {
 };
 
 bool admin_init;
+bool pubsub_init;
 bool server_init;
 bool worker_init;
diff --git a/src/core/core.c b/src/core/core.c
index a4ea7433b..90b01ea72 100644
--- a/src/core/core.c
+++ b/src/core/core.c
@@ -10,7 +10,7 @@
 #include <sysexits.h>
 
 void
-core_run(void *arg_worker)
+worker_run(void *arg_worker)
 {
     pthread_t worker, server;
     int ret;
@@ -37,3 +37,32 @@ core_run(void *arg_worker)
 error:
     exit(EX_OSERR);
 }
+
+void
+pubsub_run(void *arg_pubsub)
+{
+    pthread_t pubsub, server;
+    int ret;
+
+    if (!admin_init || !server_init || !pubsub_init) {
+        log_crit("cannot run: admin/server/pubsub have to be initialized");
+        return;
+    }
+
+    ret = pthread_create(&pubsub, NULL, core_pubsub_evloop, arg_pubsub);
+    if (ret != 0) {
+        log_crit("pthread create failed for pubsub thread: %s", strerror(ret));
+        goto error;
+    }
+
+    ret = pthread_create(&server, NULL, core_server_evloop, NULL);
+    if (ret != 0) {
+        log_crit("pthread create failed for server thread: %s", strerror(ret));
+        goto error;
+    }
+
+    core_admin_evloop();
+
+error:
+    exit(EX_OSERR);
+}
diff --git a/src/core/core.h b/src/core/core.h
index c05542c13..ce8ec3f7b 100644
--- a/src/core/core.h
+++ b/src/core/core.h
@@ -5,8 +5,10 @@
  */
 
 #include "admin/admin.h"
+#include "data/pubsub.h"
 #include "data/shared.h"
 #include "data/server.h"
 #include "data/worker.h"
 
-void core_run(void *arg_worker);
+void worker_run(void *arg_worker);
+void pubsub_run(void *arg_pubsub);
diff --git a/src/core/data/CMakeLists.txt b/src/core/data/CMakeLists.txt
index 8b3b8afbe..061e06c68 100644
--- a/src/core/data/CMakeLists.txt
+++ b/src/core/data/CMakeLists.txt
@@ -1,5 +1,6 @@
 set(SOURCE
     ${SOURCE}
+    ${CMAKE_CURRENT_SOURCE_DIR}/pubsub.c
     ${CMAKE_CURRENT_SOURCE_DIR}/server.c
     ${CMAKE_CURRENT_SOURCE_DIR}/worker.c
     PARENT_SCOPE)
diff --git a/src/core/data/worker.c b/src/core/data/worker.c
index 336fbd058..473922c31 100644
--- a/src/core/data/worker.c
+++ b/src/core/data/worker.c
@@ -291,7 +291,7 @@ _worker_evwait(void)
 void *
 core_worker_evloop(void *arg)
 {
-    post_processor = arg;
+    processor = arg;
 
     for(;;) {
         if (_worker_evwait() != CC_OK) {
diff --git a/src/server/pingserver/main.c b/src/server/pingserver/main.c
index 27c578b87..c9645fc55 100644
--- a/src/server/pingserver/main.c
+++ b/src/server/pingserver/main.c
@@ -195,7 +195,7 @@ main(int argc, char **argv)
     setup();
     option_print_all((struct option *)&setting, nopt);
 
-    core_run(&worker_processor);
+    worker_run(&worker_processor);
 
     exit(EX_OK);
 }
diff --git a/src/server/pubsub/data/CMakeLists.txt b/src/server/pubsub/data/CMakeLists.txt
index ff285f978..31a7e65f3 100644
--- a/src/server/pubsub/data/CMakeLists.txt
+++ b/src/server/pubsub/data/CMakeLists.txt
@@ -1,5 +1,4 @@
 set(SOURCE
     ${SOURCE}
     ${CMAKE_CURRENT_SOURCE_DIR}/process.c
-    ${CMAKE_CURRENT_SOURCE_DIR}/pubsub.c
     PARENT_SCOPE)
diff --git a/src/server/pubsub/data/process.c b/src/server/pubsub/data/process.c
index 9ee2d5e91..ae31eaa20 100644
--- a/src/server/pubsub/data/process.c
+++ b/src/server/pubsub/data/process.c
@@ -1,10 +1,12 @@
 #include "process.h"
 
 #include "protocol/data/redis_include.h"
-#include "pubsub.h"
+#include "storage/pubsub/listener.h"
+#include "storage/pubsub/topic.h"
 
 #include <buffer/cc_dbuf.h>
-#include <cc_debug.h>
+#include <cc_array.h>
+#include <stream/cc_sockio.h>
 
 #define PUBSUB_PROCESS_MODULE_NAME "pubsub::process"
 
@@ -13,6 +15,43 @@ command_fn command_registry[REQ_SENTINEL];
 static bool process_init = false;
 static process_metrics_st *process_metrics = NULL;
 
+static struct listener_ht *lht;
+static struct topic_ht *tht;
+
+
+/* "subscribe topic [topic ...]" */
+static void
+command_subscribe(struct response *rsp, struct request *req, struct buf_sock *s)
+{
+    struct element *el;
+    struct listener *l;
+    struct topic *t;
+    uint32_t ntopic = req->token->nelem - 1;
+
+    l = listener_ht_get(s->ch, lht);
+    if (l == NULL) {
+        l = listener_create(s->ch, s->hdl);
+        listener_ht_put(l, lht);
+    }
+
+    for (int i = 1; i < ntopic; i++) {
+        el = array_get(req->token, i);
+        if (el->type != ELEM_BULK) {
+            /* handle client error */
+        };
+
+        t = topic_ht_get(&el->bstr, tht);
+        if (t == NULL) {
+            t = topic_create(&el->bstr);
+        }
+        listener_add_topic(l, t);
+    }
+
+    rsp->type = ELEM_STR;
+    el = array_push(rsp->token);
+    el->type = ELEM_STR;
+    el->bstr = str2bstr(RSP_STR_OK);
+}
 
 void
 process_setup(process_metrics_st *metrics)
@@ -24,7 +63,8 @@ process_setup(process_metrics_st *metrics)
                  PUBSUB_PROCESS_MODULE_NAME);
     }
 
-    pubsub_setup();
+    lht = listener_ht_create(16);
+    tht = topic_ht_create(16);
 
     command_registry[REQ_SUBSCRIBE] = command_subscribe;
 
@@ -40,7 +80,8 @@ process_teardown(void)
         log_warn("%s has never been setup", PUBSUB_PROCESS_MODULE_NAME);
     }
 
-    pubsub_teardown();
+    listener_ht_destroy(&lht);
+    topic_ht_destroy(&tht);
 
     command_registry[REQ_SUBSCRIBE] = NULL;
 
diff --git a/src/server/pubsub/data/process.h b/src/server/pubsub/data/process.h
index 63ec79ddc..60a1bf5ad 100644
--- a/src/server/pubsub/data/process.h
+++ b/src/server/pubsub/data/process.h
@@ -28,3 +28,5 @@ void process_teardown(void);
 int pubsub_process_read(struct buf_sock *s);
 int pubsub_process_write(struct buf_sock *s);
 int pubsub_process_error(struct buf_sock *s);
+
+
diff --git a/src/server/pubsub/data/pubsub.c b/src/server/pubsub/data/pubsub.c
deleted file mode 100644
index cd35ed595..000000000
--- a/src/server/pubsub/data/pubsub.c
+++ /dev/null
@@ -1,59 +0,0 @@
-#include "pubsub.h"
-
-#include "protocol/data/redis_include.h"
-#include "storage/pubsub/listener.h"
-#include "storage/pubsub/topic.h"
-
-#include <cc_array.h>
-#include <stream/cc_sockio.h>
-
-static struct listener_ht *lht;
-static struct topic_ht *tht;
-
-void
-pubsub_setup(void)
-{
-    lht = listener_ht_create(16);
-    tht = topic_ht_create(16);
-}
-
-void
-pubsub_teardown(void)
-{
-    listener_ht_destroy(&lht);
-    topic_ht_destroy(&tht);
-}
-
-/* "subscribe topic [topic ...]" */
-void
-command_subscribe(struct response *rsp, struct request *req, struct buf_sock *s)
-{
-    struct element *el;
-    struct listener *l;
-    struct topic *t;
-    uint32_t ntopic = req->token->nelem - 1;
-
-    l = listener_ht_get(s->ch, lht);
-    if (l == NULL) {
-        l = listener_create(s->ch, s->hdl);
-        listener_ht_put(l, lht);
-    }
-
-    for (int i = 1; i < ntopic; i++) {
-        el = array_get(req->token, i);
-        if (el->type != ELEM_BULK) {
-            /* handle client error */
-        };
-
-        t = topic_ht_get(&el->bstr, tht);
-        if (t == NULL) {
-            t = topic_create(&el->bstr);
-        }
-        listener_add_topic(l, t);
-    }
-
-    rsp->type = ELEM_STR;
-    el = array_push(rsp->token);
-    el->type = ELEM_STR;
-    el->bstr = str2bstr(RSP_STR_OK);
-}
diff --git a/src/server/pubsub/data/pubsub.h b/src/server/pubsub/data/pubsub.h
deleted file mode 100644
index 72df15661..000000000
--- a/src/server/pubsub/data/pubsub.h
+++ /dev/null
@@ -1,12 +0,0 @@
-#pragma once
-
-struct request;
-struct response;
-struct buf_sock;
-
-void pubsub_setup(void);
-void pubsub_teardown(void);
-
-void command_subscribe(struct response *rsp, struct request *req, struct buf_sock *s);
-void command_unsubscribe(struct response *rsp, struct request *req, struct buf_sock *s);
-void command_publish(struct response *rsp, struct request *req, struct buf_sock *s);
diff --git a/src/server/pubsub/main.c b/src/server/pubsub/main.c
index 411f5fc9b..5d8e27d5d 100644
--- a/src/server/pubsub/main.c
+++ b/src/server/pubsub/main.c
@@ -13,7 +13,7 @@
 #include <sys/socket.h>
 #include <sysexits.h>
 
-struct processor worker_processor = {
+struct pubsub_processor pubsub_processor = {
     pubsub_process_read,
     pubsub_process_write,
     pubsub_process_error,
@@ -50,7 +50,9 @@ show_usage(void)
 static void
 teardown(void)
 {
-    core_teardown();
+    core_pubsub_teardown();
+    core_server_teardown();
+    core_admin_teardown();
     admin_process_teardown();
     compose_teardown();
     parse_teardown();
@@ -87,10 +89,10 @@ setup(void)
     }
 
     /* setup top-level application options */
-    if (option_bool(&setting.pubsub.daemonize)) {
+    if (option_bool(&setting.main.daemonize)) {
         daemonize();
     }
-    fname = option_str(&setting.pubsub.pid_filename);
+    fname = option_str(&setting.main.pid_filename);
     if (fname != NULL) {
         /* to get the correct pid, call create_pidfile after daemonize */
         create_pidfile(fname);
@@ -113,11 +115,12 @@ setup(void)
     compose_setup(NULL, &stats.compose_rsp);
     process_setup(&stats.process);
     admin_process_setup();
-    core_setup(&setting.admin, &setting.server, &setting.worker,
-            &stats.server, &stats.worker);
+    core_admin_setup(&setting.admin);
+    core_server_setup(&setting.server, &stats.server);
+    core_pubsub_setup(&setting.pubsub, &stats.pubsub);
 
     /* adding recurring events to maintenance/admin thread */
-    intvl = option_uint(&setting.pubsub.dlog_intvl);
+    intvl = option_uint(&setting.main.dlog_intvl);
     if (core_admin_register(intvl, debug_log_flush, NULL) == NULL) {
         log_stderr("Could not register timed event to flush debug log");
         goto error;
@@ -191,7 +194,7 @@ main(int argc, char **argv)
     setup();
     option_print_all((struct option *)&setting, nopt);
 
-    core_run(NULL, &worker_processor);
+    pubsub_run(&pubsub_processor);
 
     exit(EX_OK);
 }
diff --git a/src/server/pubsub/setting.c b/src/server/pubsub/setting.c
index 6854a0e53..29489a43a 100644
--- a/src/server/pubsub/setting.c
+++ b/src/server/pubsub/setting.c
@@ -1,10 +1,10 @@
 #include "setting.h"
 
 struct setting setting = {
-    { PUBSUB_OPTION(OPTION_INIT)    },
+    { MAIN_OPTION(OPTION_INIT)      },
     { ADMIN_OPTION(OPTION_INIT)     },
     { SERVER_OPTION(OPTION_INIT)    },
-    { WORKER_OPTION(OPTION_INIT)    },
+    { PUBSUB_OPTION(OPTION_INIT)    },
     { REQUEST_OPTION(OPTION_INIT)   },
     { RESPONSE_OPTION(OPTION_INIT)  },
     { ARRAY_OPTION(OPTION_INIT)     },
diff --git a/src/server/pubsub/setting.h b/src/server/pubsub/setting.h
index bb60374b6..23d23d976 100644
--- a/src/server/pubsub/setting.h
+++ b/src/server/pubsub/setting.h
@@ -19,22 +19,22 @@
 
 /* option related */
 /*          name            type                default description */
-#define PUBSUB_OPTION(ACTION)                                                        \
+#define MAIN_OPTION(ACTION)                                                        \
     ACTION( daemonize,      OPTION_TYPE_BOOL,   false,  "daemonize the process"        )\
     ACTION( pid_filename,   OPTION_TYPE_STR,    NULL,   "file storing the pid"         )\
     ACTION( dlog_intvl,     OPTION_TYPE_UINT,   500,    "debug log flush interval(ms)" )
 
 typedef struct {
-    PUBSUB_OPTION(OPTION_DECLARE)
-} pubsub_options_st;
+    MAIN_OPTION(OPTION_DECLARE)
+} main_options_st;
 
 struct setting {
     /* top-level */
-    pubsub_options_st       pubsub;
+    main_options_st         main;
     /* application modules */
     admin_options_st        admin;
     server_options_st       server;
-    worker_options_st       worker;
+    pubsub_options_st       pubsub;
     request_options_st      request;
     response_options_st     response;
     /* ccommon libraries */
diff --git a/src/server/pubsub/stats.c b/src/server/pubsub/stats.c
index 1c86a710a..b58d972c8 100644
--- a/src/server/pubsub/stats.c
+++ b/src/server/pubsub/stats.c
@@ -8,7 +8,7 @@ struct stats stats = {
     { REQUEST_METRIC(METRIC_INIT)       },
     { RESPONSE_METRIC(METRIC_INIT)      },
     { CORE_SERVER_METRIC(METRIC_INIT)   },
-    { CORE_WORKER_METRIC(METRIC_INIT)   },
+    { CORE_PUBSUB_METRIC(METRIC_INIT)   },
     { BUF_METRIC(METRIC_INIT)           },
     { DBUF_METRIC(METRIC_INIT)          },
     { EVENT_METRIC(METRIC_INIT)         },
diff --git a/src/server/pubsub/stats.h b/src/server/pubsub/stats.h
index 5c458fd4c..384b17bf5 100644
--- a/src/server/pubsub/stats.h
+++ b/src/server/pubsub/stats.h
@@ -23,7 +23,7 @@ struct stats {
     request_metrics_st          request;
     response_metrics_st         response;
     server_metrics_st           server;
-    worker_metrics_st           worker;
+    pubsub_metrics_st           pubsub;
     /* ccommon libraries */
     buf_metrics_st              buf;
     dbuf_metrics_st             dbuf;
diff --git a/src/server/slimcache/main.c b/src/server/slimcache/main.c
index 1d48de5b2..2291e9891 100644
--- a/src/server/slimcache/main.c
+++ b/src/server/slimcache/main.c
@@ -211,7 +211,7 @@ main(int argc, char **argv)
     setup();
     option_print_all((struct option *)&setting, nopt);
 
-    core_run(&worker_processor);
+    worker_run(&worker_processor);
 
     exit(EX_OK);
 }
diff --git a/src/server/twemcache/main.c b/src/server/twemcache/main.c
index b3b8a88f5..0f20a37d8 100644
--- a/src/server/twemcache/main.c
+++ b/src/server/twemcache/main.c
@@ -206,7 +206,7 @@ main(int argc, char **argv)
     setup();
     option_print_all((struct option *)&setting, nopt);
 
-    core_run(&worker_processor);
+    worker_run(&worker_processor);
 
     exit(EX_OK);
 }

From 26db38436435e3f7779f9e7c5faff9a353e28b40 Mon Sep 17 00:00:00 2001
From: Yao Yue <yao@twitter.com>
Date: Mon, 3 Jul 2017 17:13:59 -0700
Subject: [PATCH 5/9] rename main option

---
 src/server/pingserver/main.c    | 6 +++---
 src/server/pingserver/setting.c | 2 +-
 src/server/pingserver/setting.h | 8 ++++----
 src/server/slimcache/main.c     | 8 ++++----
 src/server/slimcache/setting.c  | 2 +-
 src/server/slimcache/setting.h  | 8 ++++----
 src/server/twemcache/main.c     | 8 ++++----
 src/server/twemcache/setting.c  | 2 +-
 src/server/twemcache/setting.h  | 8 ++++----
 9 files changed, 26 insertions(+), 26 deletions(-)

diff --git a/src/server/pingserver/main.c b/src/server/pingserver/main.c
index c9645fc55..7c0bf9712 100644
--- a/src/server/pingserver/main.c
+++ b/src/server/pingserver/main.c
@@ -93,10 +93,10 @@ setup(void)
     }
 
     /* setup top-level application options */
-    if (option_bool(&setting.pingserver.daemonize)) {
+    if (option_bool(&setting.main.daemonize)) {
         daemonize();
     }
-    fname = option_str(&setting.pingserver.pid_filename);
+    fname = option_str(&setting.main.pid_filename);
     if (fname != NULL) {
         /* to get the correct pid, call create_pidfile after daemonize */
         create_pidfile(fname);
@@ -121,7 +121,7 @@ setup(void)
     core_worker_setup(&setting.worker, &stats.worker);
 
     /* adding recurring events to maintenance/admin thread */
-    intvl = option_uint(&setting.pingserver.dlog_intvl);
+    intvl = option_uint(&setting.main.dlog_intvl);
     if (core_admin_register(intvl, debug_log_flush, NULL) == NULL) {
         log_stderr("Could not register timed event to flush debug log");
         goto error;
diff --git a/src/server/pingserver/setting.c b/src/server/pingserver/setting.c
index b699a2f4f..1976105be 100644
--- a/src/server/pingserver/setting.c
+++ b/src/server/pingserver/setting.c
@@ -1,7 +1,7 @@
 #include "setting.h"
 
 struct setting setting = {
-    { PINGSERVER_OPTION(OPTION_INIT)},
+    { MAIN_OPTION(OPTION_INIT)      },
     { ADMIN_OPTION(OPTION_INIT)     },
     { SERVER_OPTION(OPTION_INIT)    },
     { WORKER_OPTION(OPTION_INIT)    },
diff --git a/src/server/pingserver/setting.h b/src/server/pingserver/setting.h
index 035bb7400..743be265c 100644
--- a/src/server/pingserver/setting.h
+++ b/src/server/pingserver/setting.h
@@ -19,18 +19,18 @@
 
 /* option related */
 /*          name            type                default description */
-#define PINGSERVER_OPTION(ACTION)                                                        \
+#define MAIN_OPTION(ACTION)                                                             \
     ACTION( daemonize,      OPTION_TYPE_BOOL,   false,  "daemonize the process"        )\
     ACTION( pid_filename,   OPTION_TYPE_STR,    NULL,   "file storing the pid"         )\
     ACTION( dlog_intvl,     OPTION_TYPE_UINT,   500,    "debug log flush interval(ms)" )
 
 typedef struct {
-    PINGSERVER_OPTION(OPTION_DECLARE)
-} pingserver_options_st;
+    MAIN_OPTION(OPTION_DECLARE)
+} main_options_st;
 
 struct setting {
     /* top-level */
-    pingserver_options_st   pingserver;
+    main_options_st         main;
     /* application modules */
     admin_options_st        admin;
     server_options_st       server;
diff --git a/src/server/slimcache/main.c b/src/server/slimcache/main.c
index 2291e9891..3a00232e4 100644
--- a/src/server/slimcache/main.c
+++ b/src/server/slimcache/main.c
@@ -98,10 +98,10 @@ setup(void)
     }
 
     /* setup top-level application options */
-    if (option_bool(&setting.slimcache.daemonize)) {
+    if (option_bool(&setting.main.daemonize)) {
         daemonize();
     }
-    fname = option_str(&setting.slimcache.pid_filename);
+    fname = option_str(&setting.main.pid_filename);
     if (fname != NULL) {
         /* to get the correct pid, call create_pidfile after daemonize */
         create_pidfile(fname);
@@ -131,13 +131,13 @@ setup(void)
     core_worker_setup(&setting.worker, &stats.worker);
 
     /* adding recurring events to maintenance/admin thread */
-    intvl = option_uint(&setting.slimcache.dlog_intvl);
+    intvl = option_uint(&setting.main.dlog_intvl);
     if (core_admin_register(intvl, debug_log_flush, NULL) == NULL) {
         log_stderr("Could not register timed event to flush debug log");
         goto error;
     }
 
-    intvl = option_uint(&setting.slimcache.klog_intvl);
+    intvl = option_uint(&setting.main.klog_intvl);
     if (core_admin_register(intvl, klog_flush, NULL) == NULL) {
         log_error("Could not register timed event to flush command log");
         goto error;
diff --git a/src/server/slimcache/setting.c b/src/server/slimcache/setting.c
index e978d4667..09c74fa73 100644
--- a/src/server/slimcache/setting.c
+++ b/src/server/slimcache/setting.c
@@ -1,7 +1,7 @@
 #include "setting.h"
 
 struct setting setting = {
-    { SLIMCACHE_OPTION(OPTION_INIT) },
+    { MAIN_OPTION(OPTION_INIT)      },
     { ADMIN_OPTION(OPTION_INIT)     },
     { SERVER_OPTION(OPTION_INIT)    },
     { WORKER_OPTION(OPTION_INIT)    },
diff --git a/src/server/slimcache/setting.h b/src/server/slimcache/setting.h
index 5caf835fe..5f2495972 100644
--- a/src/server/slimcache/setting.h
+++ b/src/server/slimcache/setting.h
@@ -19,19 +19,19 @@
 
 /* option related */
 /*          name            type                default description */
-#define SLIMCACHE_OPTION(ACTION)                                                        \
+#define MAIN_OPTION(ACTION)                                                             \
     ACTION( daemonize,      OPTION_TYPE_BOOL,   false,  "daemonize the process"        )\
     ACTION( pid_filename,   OPTION_TYPE_STR,    NULL,   "file storing the pid"         )\
     ACTION( dlog_intvl,     OPTION_TYPE_UINT,   500,    "debug log flush interval(ms)" )\
     ACTION( klog_intvl,     OPTION_TYPE_UINT,   100,    "cmd log flush interval(ms)"   )
 
 typedef struct {
-    SLIMCACHE_OPTION(OPTION_DECLARE)
-} slimcache_options_st;
+    MAIN_OPTION(OPTION_DECLARE)
+} main_options_st;
 
 struct setting {
     /* top-level */
-    slimcache_options_st    slimcache;
+    main_options_st         main;
     /* application modules */
     admin_options_st        admin;
     server_options_st       server;
diff --git a/src/server/twemcache/main.c b/src/server/twemcache/main.c
index 0f20a37d8..4ceb9056d 100644
--- a/src/server/twemcache/main.c
+++ b/src/server/twemcache/main.c
@@ -94,10 +94,10 @@ setup(void)
     }
 
     /* setup top-level application options */
-    if (option_bool(&setting.twemcache.daemonize)) {
+    if (option_bool(&setting.main.daemonize)) {
         daemonize();
     }
-    fname = option_str(&setting.twemcache.pid_filename);
+    fname = option_str(&setting.main.pid_filename);
     if (fname != NULL) {
         /* to get the correct pid, call create_pidfile after daemonize */
         create_pidfile(fname);
@@ -127,13 +127,13 @@ setup(void)
     core_worker_setup(&setting.worker, &stats.worker);
 
     /* adding recurring events to maintenance/admin thread */
-    intvl = option_uint(&setting.twemcache.dlog_intvl);
+    intvl = option_uint(&setting.main.dlog_intvl);
     if (core_admin_register(intvl, debug_log_flush, NULL) == NULL) {
         log_stderr("Could not register timed event to flush debug log");
         goto error;
     }
 
-    intvl = option_uint(&setting.twemcache.klog_intvl);
+    intvl = option_uint(&setting.main.klog_intvl);
     if (core_admin_register(intvl, klog_flush, NULL) == NULL) {
         log_error("Could not register timed event to flush command log");
         goto error;
diff --git a/src/server/twemcache/setting.c b/src/server/twemcache/setting.c
index ceb5b0847..e05ca0838 100644
--- a/src/server/twemcache/setting.c
+++ b/src/server/twemcache/setting.c
@@ -1,7 +1,7 @@
 #include "setting.h"
 
 struct setting setting = {
-    { TWEMCACHE_OPTION(OPTION_INIT) },
+    { MAIN_OPTION(OPTION_INIT)      },
     { ADMIN_OPTION(OPTION_INIT)     },
     { SERVER_OPTION(OPTION_INIT)    },
     { WORKER_OPTION(OPTION_INIT)    },
diff --git a/src/server/twemcache/setting.h b/src/server/twemcache/setting.h
index aea199ddf..c405d0a80 100644
--- a/src/server/twemcache/setting.h
+++ b/src/server/twemcache/setting.h
@@ -17,19 +17,19 @@
 
 /* option related */
 /*          name            type                default description */
-#define TWEMCACHE_OPTION(ACTION)                                                        \
+#define MAIN_OPTION(ACTION)                                                             \
     ACTION( daemonize,      OPTION_TYPE_BOOL,   false,  "daemonize the process"        )\
     ACTION( pid_filename,   OPTION_TYPE_STR,    NULL,   "file storing the pid"         )\
     ACTION( dlog_intvl,     OPTION_TYPE_UINT,   500,    "debug log flush interval(ms)" )\
     ACTION( klog_intvl,     OPTION_TYPE_UINT,   100,    "cmd log flush interval(ms)"   )
 
 typedef struct {
-    TWEMCACHE_OPTION(OPTION_DECLARE)
-} twemcache_options_st;
+    MAIN_OPTION(OPTION_DECLARE)
+} main_options_st;
 
 struct setting {
     /* top-level */
-    twemcache_options_st    twemcache;
+    main_options_st         main;
     /* application modules */
     admin_options_st        admin;
     server_options_st       server;

From d4080f1a64dd52989360e764ad54dd3e17b340c1 Mon Sep 17 00:00:00 2001
From: Yao Yue <yao@twitter.com>
Date: Tue, 4 Jul 2017 01:47:21 -0700
Subject: [PATCH 6/9] pubsub prototype works

---
 config/pubsub.conf                   |   4 -
 src/core/data/pubsub.c               | 302 +++++++++++++++++++++++++++
 src/core/data/pubsub.h               |  53 +++++
 src/protocol/data/redis/cmd_pubsub.h |   2 +-
 src/server/pubsub/data/process.c     | 185 ++++++++++++++--
 src/server/pubsub/data/process.h     |   3 -
 src/storage/pubsub/listener.c        | 121 +++++------
 src/storage/pubsub/listener.h        |  25 +--
 src/storage/pubsub/topic.c           | 117 +++++------
 src/storage/pubsub/topic.h           |  18 +-
 10 files changed, 646 insertions(+), 184 deletions(-)
 create mode 100644 src/core/data/pubsub.c
 create mode 100644 src/core/data/pubsub.h

diff --git a/config/pubsub.conf b/config/pubsub.conf
index 90d333165..b0c646850 100644
--- a/config/pubsub.conf
+++ b/config/pubsub.conf
@@ -1,7 +1,3 @@
-# if slab profile is specified, then the profile wil be explicitly set
-# otherwise, slab profile will automatically be generated by using growth factor, slab size, and chunk size
-#slab_profile: 1024 2048 4096 8192 16384 32768 65536 131072 262144 524288
-
 debug_log_level: 6
 # debug_log_file: pubsub.log
 # debug_log_nbuf: 16384
diff --git a/src/core/data/pubsub.c b/src/core/data/pubsub.c
new file mode 100644
index 000000000..3916f22e0
--- /dev/null
+++ b/src/core/data/pubsub.c
@@ -0,0 +1,302 @@
+#include "pubsub.h"
+
+#include "shared.h"
+
+#include <time/time.h>
+
+#include <buffer/cc_buf.h>
+#include <buffer/cc_dbuf.h>
+#include <cc_debug.h>
+#include <cc_event.h>
+#include <cc_ring_array.h>
+#include <channel/cc_channel.h>
+#include <channel/cc_pipe.h>
+#include <channel/cc_tcp.h>
+
+#include <stream/cc_sockio.h>
+
+#include <sysexits.h>
+
+#define PUBSUB_MODULE_NAME "core::pubsub"
+
+pubsub_metrics_st *pubsub_metrics = NULL;
+
+static struct context context;
+struct context *ctx = &context;
+
+static channel_handler_st handlers;
+static channel_handler_st *hdl = &handlers;
+
+struct pubsub_processor *processor;
+
+static inline rstatus_i
+_pubsub_write(struct buf_sock *s)
+{
+    rstatus_i status;
+
+    log_verb("writing on buf_sock %p", s);
+
+    ASSERT(s != NULL);
+    ASSERT(s->wbuf != NULL && s->rbuf != NULL);
+
+    status = buf_tcp_write(s);
+
+    return status;
+}
+
+/* the caller only needs to check the return status of this function if
+ * it previously received a write event and wants to re-register the
+ * read event upon full, successful write.
+ */
+static inline rstatus_i
+_pubsub_event_write(struct buf_sock *s)
+{
+    rstatus_i status;
+    struct tcp_conn *c = s->ch;
+
+    status = _pubsub_write(s);
+    if (status == CC_ERETRY || status == CC_EAGAIN) { /* retry write */
+        /* by removing current masks and only listen to write event(s), we are
+         * effectively stopping processing incoming data until we can write
+         * something to the (kernel) buffer for the channel. This is sensible
+         * because either the local network or the client is backed up when
+         * kernel write buffer is full, and this allows us to propagate back
+         * pressure to the sending side.
+         */
+
+        event_del(ctx->evb, hdl->wid(c));
+        event_add_write(ctx->evb, hdl->wid(c), s);
+    } else if (status == CC_ERROR) {
+        c->state = CHANNEL_TERM;
+    }
+
+    if (processor->write(s) < 0) {
+        log_debug("handler signals channel termination");
+        s->ch->state = CHANNEL_TERM;
+        return CC_ERROR;
+    }
+
+    return status;
+}
+
+static inline void
+_pubsub_read(struct buf_sock *s)
+{
+    log_verb("reading on buf_sock %p", s);
+
+    ASSERT(s != NULL);
+    ASSERT(s->wbuf != NULL && s->rbuf != NULL);
+
+    /* TODO(kyang): consider refactoring dbuf_tcp_read and buf_tcp_read to have no return status
+       at all, since the return status is already given by the connection state */
+    buf_tcp_read(s);
+}
+
+static inline void
+pubsub_close(struct buf_sock *s)
+{
+    log_info("pubsub core close on buf_sock %p", s);
+
+    processor->error(s);
+    event_del(ctx->evb, hdl->rid(s->ch));
+    hdl->term(s->ch);
+    buf_sock_return(&s);
+}
+
+/* read event over an existing connection */
+static inline void
+_pubsub_event_read(struct buf_sock *s)
+{
+    ASSERT(s != NULL);
+
+    _pubsub_read(s);
+    if (processor->read(s) < 0) {
+        log_debug("handler signals channel termination");
+        s->ch->state = CHANNEL_TERM;
+        return;
+    }
+    if (buf_rsize(s->wbuf) > 0) {
+        log_verb("attempt to write");
+        _pubsub_event_write(s);
+    }
+}
+
+static void
+pubsub_add_conn(void)
+{
+    struct buf_sock *s;
+    char buf[RING_ARRAY_DEFAULT_CAP]; /* buffer for discarding pipe data */
+    int i;
+    rstatus_i status;
+
+    /* server pushes connection on to the ring array before writing to the pipe,
+     * therefore, we should read from the pipe first and take the connections
+     * off the ring array to match the number of bytes received.
+     *
+     * Once we move server to its own thread, it is possible that there are more
+     * connections added to the queue when we are processing, it is OK to wait
+     * for the next read event in that case.
+     */
+
+    i = pipe_recv(pipe_c, buf, RING_ARRAY_DEFAULT_CAP);
+    if (i < 0) { /* errors, do not read from ring array */
+        log_warn("not adding new connections due to pipe error");
+        return;
+    }
+
+    /* each byte in the pipe corresponds to a new connection, which we will
+     * now get from the ring array
+     */
+    for (; i > 0; --i) {
+        status = ring_array_pop(&s, conn_arr);
+        if (status != CC_OK) {
+            log_warn("event number does not match conn queue: missing %d conns",
+                    i);
+            return;
+        }
+        log_verb("Adding new buf_sock %p to pubsub thread", s);
+        s->owner = ctx;
+        s->hdl = hdl;
+        event_add_read(ctx->evb, hdl->rid(s->ch), s);
+    }
+}
+
+static void
+_pubsub_event(void *arg, uint32_t events)
+{
+    struct buf_sock *s = arg;
+    log_verb("pubsub event %06"PRIX32" on buf_sock %p", events, s);
+
+    if (s == NULL) {
+        /* event on pipe_c, new connection */
+        if (events & EVENT_READ) {
+            pubsub_add_conn();
+        } else if (events & EVENT_ERR) {
+            log_error("error event received on conn_fds pipe");
+        } else {
+            /* there should never be any write events on the pipe from pubsub */
+            NOT_REACHED();
+        }
+    } else {
+        /* event on one of the connections */
+
+        if (events & EVENT_READ) {
+            log_verb("processing pubsub read event on buf_sock %p", s);
+            INCR(pubsub_metrics, pubsub_event_read);
+            _pubsub_event_read(s);
+        } else if (events & EVENT_WRITE) {
+            log_verb("processing pubsub write event on buf_sock %p", s);
+            INCR(pubsub_metrics, pubsub_event_write);
+            if (_pubsub_event_write(s) == CC_OK) {
+                /* write backlog cleared up, re-add read event (only) */
+                event_del(ctx->evb, hdl->wid(s->ch));
+                event_add_read(ctx->evb, hdl->rid(s->ch), s);
+            }
+        } else if (events & EVENT_ERR) {
+            s->ch->state = CHANNEL_TERM;
+            INCR(pubsub_metrics, pubsub_event_error);
+        } else {
+            NOT_REACHED();
+        }
+
+        /* TODO(yao): come up with a robust policy about channel connection
+         * and pending data. Since an error can either be server (usually
+         * memory) issues or client issues (bad syntax etc), or requested (quit)
+         * it is hard to determine whether the channel should be immediately
+         * closed or not. A simplistic approach might be to always close asap,
+         * and clients should not initiate closing unless they have received all
+         * their responses. This is not as nice as the TCP half-close behavior,
+         * but simpler to implement and probably fine initially.
+         */
+        if (s->ch->state == CHANNEL_TERM || s->ch->state == CHANNEL_ERROR) {
+            pubsub_close(s);
+        }
+    }
+}
+
+void
+core_pubsub_setup(pubsub_options_st *options, pubsub_metrics_st *metrics)
+{
+    int timeout = PUBSUB_TIMEOUT;
+    int nevent = PUBSUB_NEVENT;
+
+    log_info("set up the %s module", PUBSUB_MODULE_NAME);
+
+    if (pubsub_init) {
+        log_warn("pubsub has already been setup, re-creating");
+        core_pubsub_teardown();
+    }
+
+    pubsub_metrics = metrics;
+
+    if (options != NULL) {
+        timeout = option_uint(&options->pubsub_timeout);
+        nevent = option_uint(&options->pubsub_nevent);
+    }
+
+    ctx->timeout = timeout;
+    ctx->evb = event_base_create(nevent, _pubsub_event);
+    if (ctx->evb == NULL) {
+        log_crit("failed to setup pubsub thread core; could not create event_base");
+        exit(EX_CONFIG);
+    }
+
+    hdl->accept = (channel_accept_fn)tcp_accept;
+    hdl->reject = (channel_reject_fn)tcp_reject;
+    hdl->open = (channel_open_fn)tcp_listen;
+    hdl->term = (channel_term_fn)tcp_close;
+    hdl->recv = (channel_recv_fn)tcp_recv;
+    hdl->send = (channel_send_fn)tcp_send;
+    hdl->rid = (channel_id_fn)tcp_read_id;
+    hdl->wid = (channel_id_fn)tcp_write_id;
+
+    event_add_read(ctx->evb, pipe_read_id(pipe_c), NULL);
+
+    pubsub_init = true;
+}
+
+void
+core_pubsub_teardown(void)
+{
+    log_info("tear down the %s module", PUBSUB_MODULE_NAME);
+
+    if (!pubsub_init) {
+        log_warn("%s has never been setup", PUBSUB_MODULE_NAME);
+    } else {
+        event_base_destroy(&(ctx->evb));
+    }
+    pubsub_metrics = NULL;
+    pubsub_init = false;
+}
+
+static rstatus_i
+_pubsub_evwait(void)
+{
+    int n;
+
+    n = event_wait(ctx->evb, ctx->timeout);
+    if (n < 0) {
+        return n;
+    }
+
+    INCR(pubsub_metrics, pubsub_event_loop);
+    INCR_N(pubsub_metrics, pubsub_event_total, n);
+    time_update();
+
+    return CC_OK;
+}
+
+void *
+core_pubsub_evloop(void *arg)
+{
+    processor = arg;
+
+    for(;;) {
+        if (_pubsub_evwait() != CC_OK) {
+            log_crit("pubsub core event loop exited due to failure");
+            break;
+        }
+    }
+
+    exit(1);
+}
diff --git a/src/core/data/pubsub.h b/src/core/data/pubsub.h
new file mode 100644
index 000000000..d3ebf9f1d
--- /dev/null
+++ b/src/core/data/pubsub.h
@@ -0,0 +1,53 @@
+#pragma once
+
+#include "core/context.h"
+
+#include <cc_define.h>
+#include <cc_metric.h>
+#include <cc_option.h>
+
+#define PUBSUB_TIMEOUT   100     /* in ms */
+#define PUBSUB_NEVENT    1024
+
+/*          name            type                default         description */
+#define PUBSUB_OPTION(ACTION)                                                                   \
+    ACTION( pubsub_timeout, OPTION_TYPE_UINT,   PUBSUB_TIMEOUT, "evwait timeout"               )\
+    ACTION( pubsub_nevent,  OPTION_TYPE_UINT,   PUBSUB_NEVENT,  "evwait max nevent returned"   )
+
+typedef struct {
+    PUBSUB_OPTION(OPTION_DECLARE)
+} pubsub_options_st;
+
+/*          name                    type            description */
+#define CORE_PUBSUB_METRIC(ACTION)                                                   \
+    ACTION( pubsub_event_total,     METRIC_COUNTER, "# pubsub events returned"      )\
+    ACTION( pubsub_event_loop,      METRIC_COUNTER, "# pubsub event loops returned" )\
+    ACTION( pubsub_event_read,      METRIC_COUNTER, "# pubsub core_read events"     )\
+    ACTION( pubsub_event_write,     METRIC_COUNTER, "# pubsub core_write events"    )\
+    ACTION( pubsub_event_error,     METRIC_COUNTER, "# pubsub core_error events"    )
+
+typedef struct {
+    CORE_PUBSUB_METRIC(METRIC_DECLARE)
+} pubsub_metrics_st;
+
+/*
+ * To allow the use application-specific logic in the handling of read/write
+ * events, each application is expected to implement their own versions of
+ * (post) processing functions called after the channel-level read/write is done.
+ *
+ * Applications should set and pass their instance of processor as argument
+ * to core_pubsub_evloop().
+ */
+struct buf_sock;
+typedef int (*pubsub_fn)(struct buf_sock *);
+struct pubsub_processor {
+    pubsub_fn read;
+    pubsub_fn write;
+    pubsub_fn error;
+};
+
+extern struct context *ctx;
+
+void core_pubsub_setup(pubsub_options_st *options, pubsub_metrics_st *metrics);
+void core_pubsub_teardown(void);
+void *core_pubsub_evloop(void *arg);
diff --git a/src/protocol/data/redis/cmd_pubsub.h b/src/protocol/data/redis/cmd_pubsub.h
index 446cf4ec7..b7c5600c2 100644
--- a/src/protocol/data/redis/cmd_pubsub.h
+++ b/src/protocol/data/redis/cmd_pubsub.h
@@ -1,7 +1,7 @@
 #pragma once
 
 /*          type                string          # of args */
-#define REQ_PUBSUB(ACTION)                            \
+#define REQ_PUBSUB(ACTION)                          \
     ACTION( REQ_PUBLISH,        "publish",      3  )\
     ACTION( REQ_SUBSCRIBE,      "subscribe",    -2 )\
     ACTION( REQ_UNSUBSCRIBE,    "unsubscribe",  -2 )
diff --git a/src/server/pubsub/data/process.c b/src/server/pubsub/data/process.c
index ae31eaa20..5783cb4d5 100644
--- a/src/server/pubsub/data/process.c
+++ b/src/server/pubsub/data/process.c
@@ -4,53 +4,187 @@
 #include "storage/pubsub/listener.h"
 #include "storage/pubsub/topic.h"
 
+#include "core/data/pubsub.h"
+
 #include <buffer/cc_dbuf.h>
 #include <cc_array.h>
+#include <cc_event.h>
 #include <stream/cc_sockio.h>
 
 #define PUBSUB_PROCESS_MODULE_NAME "pubsub::process"
 
+#define MESSAGE "message"
+#define SUBSCRIBE "subscribe"
+#define UNSUBSCRIBE "unsubscribe"
+
+static struct element el_message =
+    {.type = ELEM_BULK, .bstr = {sizeof(MESSAGE) - 1, MESSAGE}};
+static struct element el_subscribe =
+    {.type = ELEM_BULK, .bstr = {sizeof(SUBSCRIBE) - 1, SUBSCRIBE}};
+static struct element el_unsubscribe =
+    {.type = ELEM_BULK, .bstr = {sizeof(UNSUBSCRIBE) - 1, UNSUBSCRIBE}};
+
+
+typedef void (* command_fn)(struct response *, struct request *, struct buf_sock *);
 command_fn command_registry[REQ_SENTINEL];
 
 static bool process_init = false;
 static process_metrics_st *process_metrics = NULL;
 
-static struct listener_ht *lht;
-static struct topic_ht *tht;
 
+/* "publish topic msg"
+ * reply: # listeners subscribed
+ */
+static void
+command_publish(struct response *rsp, struct request *req, struct buf_sock *s)
+{
+    struct element *el_t, *el_m; /* topic & message */
+    struct element el_r = {.type = ELEM_INT}; /* reply, an integer */
+    struct listener *l;
+    struct topic *t;
+    uint32_t nsub = 0;
+
+    el_t = array_get(req->token, 1);
+    if (el_t->type != ELEM_BULK) {
+        /* handle error */
+    };
+    el_m = array_get(req->token, 2);
+    if (el_m->type != ELEM_BULK) {
+        /* handle error */
+    };
+
+    log_verb("publish from buf_sock %p", s);
+
+    l = listener_get(s);
+    if (l != NULL) {
+        log_error("found listener at %p: subscriber cannot publish", l);
+        /* handle error */
+    }
+
+    t = topic_get(&el_t->bstr);
+    if (t == NULL) {
+        log_verb("no listener on topic %.*s, ignore", el_t->bstr.len,
+                el_t->bstr.data);
+    } else { /* copy message to listener's buffer, this should be optimized later */
+        struct index_node *node;
+
+        nsub = t->nsub;
+        TAILQ_FOREACH(node, t->idx, i_tqe) {
+            l = (struct listener *)node->obj;
+            struct buf_sock *ss = l->s;
+            compose_array_header(&ss->wbuf, 3);
+            compose_element(&ss->wbuf, &el_message);
+            compose_element(&ss->wbuf, el_t);
+            compose_element(&ss->wbuf, el_m);
+            /* register listener's fd for write event */
+            event_add_write(ctx->evb, ss->hdl->wid(ss->ch), ss);
+        }
+    }
+
+    el_r.num = nsub;
+    compose_element(&s->wbuf, &el_r);
+}
 
-/* "subscribe topic [topic ...]" */
+/* "subscribe topic [topic ...]"
+ * no reply
+ */
 static void
 command_subscribe(struct response *rsp, struct request *req, struct buf_sock *s)
 {
     struct element *el;
+    struct element el_s = {.type = ELEM_INT}; /* # topics subscribed, an integer */
     struct listener *l;
     struct topic *t;
     uint32_t ntopic = req->token->nelem - 1;
 
-    l = listener_ht_get(s->ch, lht);
+    log_verb("subscribe buf_sock %p from topics", s);
+
+    l = listener_get(s);
     if (l == NULL) {
-        l = listener_create(s->ch, s->hdl);
-        listener_ht_put(l, lht);
+        log_verb("create new listener for %p", s);
+        l = listener_create(s);
+        listener_put(l);
     }
 
-    for (int i = 1; i < ntopic; i++) {
+    for (int i = 1; i <= ntopic; i++) {
         el = array_get(req->token, i);
         if (el->type != ELEM_BULK) {
-            /* handle client error */
+            /* handle error */
         };
 
-        t = topic_ht_get(&el->bstr, tht);
+        t = topic_get(&el->bstr);
         if (t == NULL) {
+            log_verb("creating topic %.*s", el->bstr.len, el->bstr.data);
             t = topic_create(&el->bstr);
+            /* handle error */
+            topic_put(t);
+        }
+
+        log_verb("subscribing to topic %.*s", el->bstr.len, el->bstr.data);
+        if (!topic_add_listener(t, l)) {
+            log_debug("listener not added");
+        }
+        if (!listener_add_topic(l, t)) {
+            log_debug("topic not added");
         }
-        listener_add_topic(l, t);
+
+        el_s.num = l->ntopic;
+        compose_array_header(&s->wbuf, 3);
+        compose_element(&s->wbuf, &el_subscribe);
+        compose_element(&s->wbuf, el);
+        compose_element(&s->wbuf, &el_s);
     }
+}
 
-    rsp->type = ELEM_STR;
-    el = array_push(rsp->token);
-    el->type = ELEM_STR;
-    el->bstr = str2bstr(RSP_STR_OK);
+/* "unsubscribe topic [topic ...]"
+ * no reply
+ * (not supporting unsubscribe from all right now)
+ */
+static void
+command_unsubscribe(struct response *rsp, struct request *req, struct buf_sock *s)
+{
+    struct element *el;
+    struct element el_s = {.type = ELEM_INT}; /* # topics subscribed, an integer */
+    struct listener *l;
+    struct topic *t;
+    uint32_t ntopic = req->token->nelem - 1;
+
+    log_verb("unsubscribe buf_sock %p from topics", s);
+
+    l = listener_get(s);
+    if (l == NULL) {
+        log_info("listener not found for %p", s);
+        return;
+    }
+
+    for (int i = 1; i <= ntopic; i++) {
+        el = array_get(req->token, i);
+        if (el->type != ELEM_BULK) {
+            /* handle error */
+        };
+
+        t = topic_get(&el->bstr);
+        if (t == NULL) {
+            log_debug("topic %.*s does not exist", el->bstr.len, el->bstr.data);
+            continue;
+        }
+
+        log_verb("unsubscribing from topic %.*s", el->bstr.len, el->bstr.data);
+        listener_del_topic(l, t);
+        topic_del_listener(t, l);
+
+        if (t->nsub == 0) { /* remove topic that nobody listens to */
+            log_verb("deleting topic %.*s", el->bstr.len, el->bstr.data);
+            topic_delete(&t->name);
+            topic_destroy(&t);
+        }
+
+        el_s.num = l->ntopic;
+        compose_array_header(&s->wbuf, 3);
+        compose_element(&s->wbuf, &el_unsubscribe);
+        compose_element(&s->wbuf, el);
+        compose_element(&s->wbuf, &el_s);
+    }
 }
 
 void
@@ -63,10 +197,12 @@ process_setup(process_metrics_st *metrics)
                  PUBSUB_PROCESS_MODULE_NAME);
     }
 
-    lht = listener_ht_create(16);
-    tht = topic_ht_create(16);
+    listener_setup(16);
+    topic_setup(16);
 
+    command_registry[REQ_PUBLISH] = command_publish;
     command_registry[REQ_SUBSCRIBE] = command_subscribe;
+    command_registry[REQ_UNSUBSCRIBE] = command_unsubscribe;
 
     process_metrics = metrics;
     process_init = true;
@@ -80,10 +216,12 @@ process_teardown(void)
         log_warn("%s has never been setup", PUBSUB_PROCESS_MODULE_NAME);
     }
 
-    listener_ht_destroy(&lht);
-    topic_ht_destroy(&tht);
+    listener_teardown();
+    topic_teardown();
 
+    command_registry[REQ_PUBLISH] = NULL;
     command_registry[REQ_SUBSCRIBE] = NULL;
+    command_registry[REQ_UNSUBSCRIBE] = NULL;
 
     process_metrics = NULL;
     process_init = false;
@@ -99,6 +237,7 @@ process_request_sock(struct response *rsp, struct request *req, struct buf_sock
         /* return error */
     }
 
+    log_verb("processing request type %d", req->type);
     command_registry[req->type](rsp, req, s);
 }
 
@@ -177,11 +316,17 @@ pubsub_process_write(struct buf_sock *s)
 int
 pubsub_process_error(struct buf_sock *s)
 {
-    struct request *req;
-    struct response *rsp;
+    struct listener *l;
 
     log_verb("post-error processing");
 
+    l = listener_get(s);
+    if (l != NULL) {
+        listener_delete(s);
+        /* TODO: unsubscribe automatically from all */
+        listener_destroy(&l);
+    }
+
     /* normalize buffer size */
     buf_reset(s->rbuf);
     dbuf_shrink(&s->rbuf);
diff --git a/src/server/pubsub/data/process.h b/src/server/pubsub/data/process.h
index 60a1bf5ad..c7d49912f 100644
--- a/src/server/pubsub/data/process.h
+++ b/src/server/pubsub/data/process.h
@@ -19,9 +19,6 @@ typedef struct {
     PROCESS_METRIC(METRIC_DECLARE)
 } process_metrics_st;
 
-typedef void (* command_fn)(struct response *, struct request *, struct buf_sock *);
-extern command_fn command_registry[REQ_SENTINEL];
-
 void process_setup(process_metrics_st *metrics);
 void process_teardown(void);
 
diff --git a/src/storage/pubsub/listener.c b/src/storage/pubsub/listener.c
index ed103872c..4a0679be0 100644
--- a/src/storage/pubsub/listener.c
+++ b/src/storage/pubsub/listener.c
@@ -5,78 +5,37 @@
 #include <cc_queue.h>
 #include <hash/cc_lookup3.h>
 
-static struct listener_slh *
-_ht_alloc(uint32_t nentry)
-{
-    struct listener_slh *table;
-    uint32_t i;
+#include <sysexits.h>
 
-    table = cc_alloc(sizeof(*table) * nentry);
-
-    if (table != NULL) {
-        for (i = 0; i < nentry; ++i) {
-            SLIST_INIT(&table[i]);
-        }
-    }
-
-    return table;
-}
-
-struct listener_ht *
-listener_ht_create(uint32_t hash_power)
-{
-    struct listener_ht *ht;
-    uint32_t nentry;
-
-    ASSERT(hash_power > 0);
+SLIST_HEAD(listener_slh, listener);
 
-    ht = cc_alloc(sizeof(struct listener_ht));
-    if (ht == NULL) {
-        return NULL;
-    }
-
-    ht->hash_power = hash_power;
-    ht->nlistener = 0;
-    nentry = HASHSIZE(ht->hash_power);
-    ht->table = _ht_alloc(nentry);
-    if (ht->table == NULL) {
-        cc_free(ht);
-        return NULL;
-    }
-
-    return ht;
-}
-
-void listener_ht_destroy(struct listener_ht **ht)
-{
-    ASSERT(ht != NULL);
-
-    if (*ht != NULL && (*ht)->table != NULL) {
-        cc_free((*ht)->table);
-    }
+struct listener_ht {
+    struct listener_slh *table;
+    uint32_t nlistener;
+    uint32_t hash_power;
+};
 
-    cc_free(*ht);
-    *ht = NULL;
-}
+static struct listener_ht hashtable;
+static struct listener_ht *ht = &hashtable;
 
 
 static struct listener_slh *
-_get_bucket(const channel_p ch, struct listener_ht *ht)
+_get_bucket(const struct buf_sock *s)
 {
     /* use the _address_ of the channel to hash */
-    uint32_t hval = hash_lookup3((char *)&ch, sizeof(channel_p), 0);
+    uint32_t hval = hash_lookup3((char *)s, sizeof(struct buf_sock *), 0);
     return &(ht->table[hval & HASHMASK(ht->hash_power)]);
 }
 
 struct listener *
-listener_ht_get(const channel_p ch, struct listener_ht *ht)
+listener_get(const struct buf_sock *s)
 {
     struct listener_slh *bucket;
     struct listener *l;
 
-    bucket = _get_bucket(ch, ht);
+    bucket = _get_bucket(s);
     for (l = SLIST_FIRST(bucket); l != NULL; l = SLIST_NEXT(l, l_sle)) {
-        if (l->ch == ch) {
+        if (l->s == s) {
             return l;
         }
     }
@@ -85,28 +44,28 @@ listener_ht_get(const channel_p ch, struct listener_ht *ht)
 }
 
 void
-listener_ht_put(const struct listener *l, struct listener_ht *ht)
+listener_put(const struct listener *l)
 {
     struct listener_slh *bucket;
 
-    ASSERT(listener_ht_get(l->ch, ht) == NULL);
+    ASSERT(listener_get(l->s) == NULL);
 
-    bucket = _get_bucket(l->ch, ht);
+    bucket = _get_bucket(l->s);
     SLIST_INSERT_HEAD(bucket, (struct listener *)l, l_sle);
 
     ht->nlistener++;
 }
 
 void
-listener_ht_delete(const channel_p ch, struct listener_ht *ht)
+listener_delete(const struct buf_sock *s)
 {
     struct listener_slh *bucket;
     struct listener *l, *prev;
 
-    bucket = _get_bucket(ch, ht);
+    bucket = _get_bucket(s);
     for (prev = NULL, l = SLIST_FIRST(bucket); l != NULL;
         prev = l, l = SLIST_NEXT(l, l_sle)) {
-        if (l->ch == ch) {
+        if (l->s == s) {
             break;
         }
     }
@@ -122,7 +81,7 @@ listener_ht_delete(const channel_p ch, struct listener_ht *ht)
 
 
 struct listener *
-listener_create(channel_p ch, channel_handler_st *handler)
+listener_create(struct buf_sock *s)
 {
     struct listener *l;
 
@@ -138,8 +97,7 @@ listener_create(channel_p ch, channel_handler_st *handler)
     }
 
     listener_reset(l);
-    l->ch = ch;
-    l->handler = handler;
+    l->s = s;
 
     return l;
 }
@@ -166,8 +124,7 @@ listener_destroy(struct listener **l)
 void
 listener_reset(struct listener *l)
 {
-    l->ch = NULL;
-    l->handler = NULL;
+    l->s = NULL;
     l->ntopic = 0;
     TAILQ_INIT(l->idx);
 }
@@ -188,6 +145,7 @@ listener_add_topic(struct listener *l, const struct topic *t)
 
     node = cc_alloc(sizeof(struct index_node));
     if (node == NULL) {
+        log_error("cannot add topic: out of memory");
         return false;
     }
     node->obj = (struct topic *)t;
@@ -216,3 +174,34 @@ listener_del_topic(struct listener *l, const struct topic *t)
     TAILQ_REMOVE(l->idx, node, i_tqe);
     l->ntopic--;
 }
+
+void
+listener_setup(uint32_t hash_power)
+{
+    uint32_t i, nentry;
+    struct listener_slh *table;
+
+    ASSERT(hash_power > 0);
+
+    ht->hash_power = hash_power;
+    ht->nlistener = 0;
+    nentry = HASHSIZE(ht->hash_power);
+
+    table = cc_alloc(sizeof(*table) * nentry);
+    if (table == NULL) {
+        log_crit("listener setup failed: OOM");
+        exit(EX_CONFIG);
+    }
+    ht->table = table;
+
+    for (i = 0; i < nentry; ++i) {
+        SLIST_INIT(&table[i]);
+    }
+}
+
+void listener_teardown(void)
+{
+    if (ht->table != NULL) {
+        cc_free(ht->table);
+    }
+}
diff --git a/src/storage/pubsub/listener.h b/src/storage/pubsub/listener.h
index 872ade1c2..e0273d229 100644
--- a/src/storage/pubsub/listener.h
+++ b/src/storage/pubsub/listener.h
@@ -2,7 +2,7 @@
 
 #include "index.h"
 
-#include <channel/cc_channel.h>
+#include <stream/cc_sockio.h>
 
 /*
  * a listener is a client that has subscribed to at least one channel
@@ -10,30 +10,21 @@
 struct listener {
     SLIST_ENTRY(listener)   l_sle;
 
-    channel_p               ch;
-    channel_handler_st      *handler;
+    struct buf_sock         *s;
     uint32_t                ntopic;
     struct index_tqh        *idx; /* index of all topics */
 };
 
-SLIST_HEAD(listener_slh, listener);
-
-struct listener_ht {
-    struct listener_slh *table;
-    uint32_t nlistener;
-    uint32_t hash_power;
-};
-
 struct topic;
 
-struct listener_ht *listener_ht_create(uint32_t hash_power);
-void listener_ht_destroy(struct listener_ht **ht);
+void listener_setup(uint32_t hash_power);
+void listener_teardown(void);
 
-struct listener *listener_ht_get(const channel_p ch, struct listener_ht *ht);
-void listener_ht_put(const struct listener *l, struct listener_ht *ht);
-void listener_ht_delete(const channel_p ch, struct listener_ht *ht);
+struct listener *listener_get(const struct buf_sock *s);
+void listener_put(const struct listener *l);
+void listener_delete(const struct buf_sock *s);
 
-struct listener *listener_create(channel_p ch, channel_handler_st *handler);
+struct listener *listener_create(struct buf_sock *s);
 void listener_destroy(struct listener **l);
 void listener_reset(struct listener *l);
 
diff --git a/src/storage/pubsub/topic.c b/src/storage/pubsub/topic.c
index 77b40ff15..d121fbd03 100644
--- a/src/storage/pubsub/topic.c
+++ b/src/storage/pubsub/topic.c
@@ -5,64 +5,22 @@
 #include <cc_queue.h>
 #include <hash/cc_lookup3.h>
 
-static struct topic_slh *
-_ht_alloc(uint32_t nentry)
-{
-    struct topic_slh *table;
-    uint32_t i;
-
-    table = cc_alloc(sizeof(*table) * nentry);
-
-    if (table != NULL) {
-        for (i = 0; i < nentry; ++i) {
-            SLIST_INIT(&table[i]);
-        }
-    }
-
-    return table;
-}
-
-struct topic_ht *
-topic_ht_create(uint32_t hash_power)
-{
-    struct topic_ht *ht;
-    uint32_t nentry;
-
-    ASSERT(hash_power > 0);
+#include <sysexits.h>
 
-    ht = cc_alloc(sizeof(struct topic_ht));
-    if (ht == NULL) {
-        return NULL;
-    }
-
-    ht->hash_power = hash_power;
-    ht->ntopic = 0;
-    nentry = HASHSIZE(ht->hash_power);
-    ht->table = _ht_alloc(nentry);
-    if (ht->table == NULL) {
-        cc_free(ht);
-        return NULL;
-    }
-
-    return ht;
-}
-
-void
-topic_ht_destroy(struct topic_ht **ht)
-{
-    ASSERT(ht != NULL);
+SLIST_HEAD(topic_slh, topic);
 
-    if (*ht != NULL && (*ht)->table != NULL) {
-        cc_free((*ht)->table);
-    }
+struct topic_ht {
+    struct topic_slh *table;
+    uint32_t ntopic;
+    uint32_t hash_power;
+};
 
-    cc_free(*ht);
-    *ht = NULL;
-}
+static struct topic_ht hashtable;
+static struct topic_ht *ht = &hashtable;
 
 
 static struct topic_slh *
-_get_bucket(const struct bstring *name, struct topic_ht *ht)
+_get_bucket(const struct bstring *name)
 {
     /* use the _address_ of the channel to hash */
     uint32_t hval = hash_lookup3(name->data, name->len, 0);
@@ -70,13 +28,14 @@ _get_bucket(const struct bstring *name, struct topic_ht *ht)
 }
 
 struct topic *
-topic_ht_get(const struct bstring *name, struct topic_ht *ht)
+topic_get(const struct bstring *name)
 {
     struct topic_slh *bucket;
     struct topic *t;
 
-    bucket = _get_bucket(name, ht);
+    bucket = _get_bucket(name);
     for (t = SLIST_FIRST(bucket); t != NULL; t = SLIST_NEXT(t, t_sle)) {
+        log_verb("current topic name (len %d) is: %.*s", t->name.len, t->name.len, t->name.data);
         if (bstring_compare(&t->name, name) == 0) {
             return t;
         }
@@ -86,25 +45,25 @@ topic_ht_get(const struct bstring *name, struct topic_ht *ht)
 }
 
 void
-topic_ht_put(const struct topic *t, struct topic_ht *ht)
+topic_put(const struct topic *t)
 {
     struct topic_slh *bucket;
 
-    ASSERT(topic_ht_get(&t->name, ht) == NULL);
+    ASSERT(topic_get(&t->name) == NULL);
 
-    bucket = _get_bucket(&t->name, ht);
+    bucket = _get_bucket(&t->name);
     SLIST_INSERT_HEAD(bucket, (struct topic *)t, t_sle);
 
     ht->ntopic++;
 }
 
 void
-topic_ht_delete(const struct bstring *name, struct topic_ht *ht)
+topic_delete(const struct bstring *name)
 {
     struct topic_slh *bucket;
     struct topic *t, *prev;
 
-    bucket = _get_bucket(name, ht);
+    bucket = _get_bucket(name);
     for (prev = NULL, t = SLIST_FIRST(bucket); t != NULL;
         prev = t, t = SLIST_NEXT(t, t_sle)) {
         if (bstring_compare(&t->name, name) == 0) {
@@ -139,7 +98,13 @@ topic_create(const struct bstring *name)
     }
 
     topic_reset(t);
-    t->name = *name;
+    t->name.len = name->len;
+    t->name.data = cc_alloc(t->name.len);
+    if (t->name.data == NULL) {
+        cc_free(t);
+        return NULL;
+    }
+    cc_memcpy(t->name.data, name->data, name->len);
 
     return t;
 }
@@ -158,6 +123,7 @@ topic_destroy(struct topic **t)
         cc_free(curr);
     }
     cc_free(idx);
+    cc_free((*t)->name.data);
 
     cc_free(*t);
     *t = NULL;
@@ -216,3 +182,34 @@ topic_del_listener(struct topic *t, const struct listener *l)
     t->nsub--;
     cc_free(node);
 }
+
+void
+topic_setup(uint32_t hash_power)
+{
+    uint32_t i, nentry;
+    struct topic_slh *table;
+
+    ASSERT(hash_power > 0);
+
+    ht->hash_power = hash_power;
+    ht->ntopic = 0;
+    nentry = HASHSIZE(ht->hash_power);
+
+    table = cc_alloc(sizeof(*table) * nentry);
+    if (table == NULL) {
+        log_crit("topic setup failed: OOM");
+        exit(EX_CONFIG);
+    }
+    ht->table = table;
+
+    for (i = 0; i < nentry; ++i) {
+        SLIST_INIT(&table[i]);
+    }
+}
+
+void topic_teardown(void)
+{
+    if (ht->table != NULL) {
+        cc_free(ht->table);
+    }
+}
diff --git a/src/storage/pubsub/topic.h b/src/storage/pubsub/topic.h
index f9ad7d9f2..d8370fea7 100644
--- a/src/storage/pubsub/topic.h
+++ b/src/storage/pubsub/topic.h
@@ -16,22 +16,14 @@ struct topic {
     struct index_tqh    *idx; /* index of all listeners */
 };
 
-SLIST_HEAD(topic_slh, topic);
-
-struct topic_ht {
-    struct topic_slh *table;
-    uint32_t ntopic;
-    uint32_t hash_power;
-};
-
 struct listener;
 
-struct topic_ht *topic_ht_create(uint32_t hash_power);
-void topic_ht_destroy(struct topic_ht **ht);
+void topic_setup(uint32_t hash_power);
+void topic_teardown(void);
 
-struct topic *topic_ht_get(const struct bstring *name, struct topic_ht *ht);
-void topic_ht_put(const struct topic *t, struct topic_ht *ht);
-void topic_ht_delete(const struct bstring *name, struct topic_ht *ht);
+struct topic *topic_get(const struct bstring *name);
+void topic_put(const struct topic *t);
+void topic_delete(const struct bstring *name);
 
 struct topic *topic_create(const struct bstring *name);
 void topic_destroy(struct topic **t);

From 94ad99d01999f436cb333a0dc3555c3af85e32ad Mon Sep 17 00:00:00 2001
From: Yao Yue <yao@twitter.com>
Date: Mon, 10 Jul 2017 22:21:10 -0700
Subject: [PATCH 7/9] address the simpler issues called out by seppo0010

---
 src/core/data/pubsub.c            | 2 +-
 src/server/pubsub/admin/process.c | 2 ++
 src/server/pubsub/data/process.c  | 4 +---
 src/storage/pubsub/listener.c     | 5 +++++
 src/storage/pubsub/topic.c        | 5 +++++
 5 files changed, 14 insertions(+), 4 deletions(-)

diff --git a/src/core/data/pubsub.c b/src/core/data/pubsub.c
index 3916f22e0..2f77d0e88 100644
--- a/src/core/data/pubsub.c
+++ b/src/core/data/pubsub.c
@@ -72,7 +72,7 @@ _pubsub_event_write(struct buf_sock *s)
 
     if (processor->write(s) < 0) {
         log_debug("handler signals channel termination");
-        s->ch->state = CHANNEL_TERM;
+        c->state = CHANNEL_TERM;
         return CC_ERROR;
     }
 
diff --git a/src/server/pubsub/admin/process.c b/src/server/pubsub/admin/process.c
index 00f36d060..43fb64f89 100644
--- a/src/server/pubsub/admin/process.c
+++ b/src/server/pubsub/admin/process.c
@@ -38,6 +38,8 @@ admin_process_teardown(void)
         log_warn("%s has never been setup", PUBSUB_ADMIN_MODULE_NAME);
     }
 
+    cc_free(buf);
+
     admin_init = false;
 }
 
diff --git a/src/server/pubsub/data/process.c b/src/server/pubsub/data/process.c
index 5783cb4d5..08f6b63ef 100644
--- a/src/server/pubsub/data/process.c
+++ b/src/server/pubsub/data/process.c
@@ -39,7 +39,6 @@ static void
 command_publish(struct response *rsp, struct request *req, struct buf_sock *s)
 {
     struct element *el_t, *el_m; /* topic & message */
-    struct element el_r = {.type = ELEM_INT}; /* reply, an integer */
     struct listener *l;
     struct topic *t;
     uint32_t nsub = 0;
@@ -81,8 +80,7 @@ command_publish(struct response *rsp, struct request *req, struct buf_sock *s)
         }
     }
 
-    el_r.num = nsub;
-    compose_element(&s->wbuf, &el_r);
+    compose_element(&s->wbuf, &(struct element){.type = ELEM_INT, .num = nsub});
 }
 
 /* "subscribe topic [topic ...]"
diff --git a/src/storage/pubsub/listener.c b/src/storage/pubsub/listener.c
index 4a0679be0..f8d5661d6 100644
--- a/src/storage/pubsub/listener.c
+++ b/src/storage/pubsub/listener.c
@@ -65,11 +65,16 @@ listener_delete(const struct buf_sock *s)
     bucket = _get_bucket(s);
     for (prev = NULL, l = SLIST_FIRST(bucket); l != NULL;
         prev = l, l = SLIST_NEXT(l, l_sle)) {
+        if (l == NULL) {
+            log_debug("listener not found for buf_sock %p", s);
+            return;
+        }
         if (l->s == s) {
             break;
         }
     }
 
+
     if (prev == NULL) {
         SLIST_REMOVE_HEAD(bucket, l_sle);
     } else {
diff --git a/src/storage/pubsub/topic.c b/src/storage/pubsub/topic.c
index d121fbd03..f6f4e1b4b 100644
--- a/src/storage/pubsub/topic.c
+++ b/src/storage/pubsub/topic.c
@@ -66,6 +66,10 @@ topic_delete(const struct bstring *name)
     bucket = _get_bucket(name);
     for (prev = NULL, t = SLIST_FIRST(bucket); t != NULL;
         prev = t, t = SLIST_NEXT(t, t_sle)) {
+        if (t == NULL) {
+            log_debug("topic not found for %.*s", name->len, name->data);
+            return;
+        }
         if (bstring_compare(&t->name, name) == 0) {
             break;
         }
@@ -101,6 +105,7 @@ topic_create(const struct bstring *name)
     t->name.len = name->len;
     t->name.data = cc_alloc(t->name.len);
     if (t->name.data == NULL) {
+        cc_free(t->idx);
         cc_free(t);
         return NULL;
     }

From f84b82aa2cf74c83d0f958d7d9cfd1ebae5bc027 Mon Sep 17 00:00:00 2001
From: Yao Yue <yao@twitter.com>
Date: Tue, 11 Jul 2017 01:19:47 -0700
Subject: [PATCH 8/9] simplify topic/listener APIs by hiding object
 create/destroy

---
 src/server/pubsub/data/process.c |  20 +----
 src/storage/pubsub/listener.c    | 131 ++++++++++++++++------------
 src/storage/pubsub/listener.h    |   6 +-
 src/storage/pubsub/topic.c       | 142 +++++++++++++++++--------------
 src/storage/pubsub/topic.h       |   6 +-
 5 files changed, 160 insertions(+), 145 deletions(-)

diff --git a/src/server/pubsub/data/process.c b/src/server/pubsub/data/process.c
index 08f6b63ef..1099387a8 100644
--- a/src/server/pubsub/data/process.c
+++ b/src/server/pubsub/data/process.c
@@ -99,9 +99,7 @@ command_subscribe(struct response *rsp, struct request *req, struct buf_sock *s)
 
     l = listener_get(s);
     if (l == NULL) {
-        log_verb("create new listener for %p", s);
-        l = listener_create(s);
-        listener_put(l);
+        l = listener_add(s);
     }
 
     for (int i = 1; i <= ntopic; i++) {
@@ -112,10 +110,8 @@ command_subscribe(struct response *rsp, struct request *req, struct buf_sock *s)
 
         t = topic_get(&el->bstr);
         if (t == NULL) {
-            log_verb("creating topic %.*s", el->bstr.len, el->bstr.data);
-            t = topic_create(&el->bstr);
+            t = topic_add(&el->bstr);
             /* handle error */
-            topic_put(t);
         }
 
         log_verb("subscribing to topic %.*s", el->bstr.len, el->bstr.data);
@@ -172,9 +168,7 @@ command_unsubscribe(struct response *rsp, struct request *req, struct buf_sock *
         topic_del_listener(t, l);
 
         if (t->nsub == 0) { /* remove topic that nobody listens to */
-            log_verb("deleting topic %.*s", el->bstr.len, el->bstr.data);
             topic_delete(&t->name);
-            topic_destroy(&t);
         }
 
         el_s.num = l->ntopic;
@@ -314,16 +308,10 @@ pubsub_process_write(struct buf_sock *s)
 int
 pubsub_process_error(struct buf_sock *s)
 {
-    struct listener *l;
-
     log_verb("post-error processing");
 
-    l = listener_get(s);
-    if (l != NULL) {
-        listener_delete(s);
-        /* TODO: unsubscribe automatically from all */
-        listener_destroy(&l);
-    }
+    listener_delete(s);
+    /* TODO: unsubscribe automatically from all */
 
     /* normalize buffer size */
     buf_reset(s->rbuf);
diff --git a/src/storage/pubsub/listener.c b/src/storage/pubsub/listener.c
index f8d5661d6..06fdadfc4 100644
--- a/src/storage/pubsub/listener.c
+++ b/src/storage/pubsub/listener.c
@@ -27,6 +27,60 @@ _get_bucket(const struct buf_sock *s)
     return &(ht->table[hval & HASHMASK(ht->hash_power)]);
 }
 
+static void
+_listener_reset(struct listener *l)
+{
+    l->s = NULL;
+    l->ntopic = 0;
+    TAILQ_INIT(l->idx);
+}
+
+static struct listener *
+_listener_create(const struct buf_sock *s)
+{
+    struct listener *l;
+
+    log_verb("create new listener for buf_sock %p", s);
+
+    l = cc_alloc(sizeof(struct listener));
+    if (l == NULL) {
+        return NULL;
+    }
+
+    l->idx = cc_alloc(sizeof(struct index_tqh));
+    if (l->idx == NULL) {
+        cc_free(l);
+        return NULL;
+    }
+
+    _listener_reset(l);
+    l->s = (struct buf_sock *)s;
+
+    return l;
+}
+
+static void
+_listener_destroy(struct listener **l)
+{
+    ASSERT(l != NULL && *l != NULL);
+
+    log_verb("destroy listener %p", *l);
+
+    struct index_node *curr, *next;
+    struct index_tqh *idx = (*l)->idx;
+
+    /* delete all elements of the index */
+    TAILQ_FOREACH_SAFE(curr, idx, i_tqe, next) {
+        TAILQ_REMOVE(idx, curr, i_tqe);
+        cc_free(curr);
+    }
+    cc_free(idx);
+
+    cc_free(*l);
+    *l = NULL;
+}
+
+
 struct listener *
 listener_get(const struct buf_sock *s)
 {
@@ -40,20 +94,27 @@ listener_get(const struct buf_sock *s)
         }
     }
 
+    log_verb("listener not found for buf_sock", s);
     return NULL;
 }
 
-void
-listener_put(const struct listener *l)
+struct listener *
+listener_add(const struct buf_sock *s)
 {
+    ASSERT(listener_get(s) == NULL);
+
     struct listener_slh *bucket;
+    struct listener *l = _listener_create(s);
 
-    ASSERT(listener_get(l->s) == NULL);
+    log_verb("add listener %p for buf_sock %p", l, s);
 
-    bucket = _get_bucket(l->s);
-    SLIST_INSERT_HEAD(bucket, (struct listener *)l, l_sle);
+    bucket = _get_bucket(s);
+    SLIST_INSERT_HEAD(bucket, l, l_sle);
 
     ht->nlistener++;
+    log_verb("total listeners: %"PRIu32, ht->nlistener);
+
+    return l;
 }
 
 void
@@ -74,64 +135,15 @@ listener_delete(const struct buf_sock *s)
         }
     }
 
-
     if (prev == NULL) {
         SLIST_REMOVE_HEAD(bucket, l_sle);
     } else {
         SLIST_REMOVE_AFTER(prev, l_sle);
     }
 
+    _listener_destroy(&l);
     --(ht->nlistener);
-}
-
-
-struct listener *
-listener_create(struct buf_sock *s)
-{
-    struct listener *l;
-
-    l = cc_alloc(sizeof(struct listener));
-    if (l == NULL) {
-        return NULL;
-    }
-
-    l->idx = cc_alloc(sizeof(struct index_tqh));
-    if (l->idx == NULL) {
-        cc_free(l);
-        return NULL;
-    }
-
-    listener_reset(l);
-    l->s = s;
-
-    return l;
-}
-
-void
-listener_destroy(struct listener **l)
-{
-    ASSERT(l != NULL && *l != NULL);
-
-    struct index_node *curr, *next;
-    struct index_tqh *idx = (*l)->idx;
-
-    /* delete all elements of the index */
-    TAILQ_FOREACH_SAFE(curr, idx, i_tqe, next) {
-        TAILQ_REMOVE(idx, curr, i_tqe);
-        cc_free(curr);
-    }
-    cc_free(idx);
-
-    cc_free(*l);
-    *l = NULL;
-}
-
-void
-listener_reset(struct listener *l)
-{
-    l->s = NULL;
-    l->ntopic = 0;
-    TAILQ_INIT(l->idx);
+    log_verb("total listeners: %"PRIu32, ht->nlistener);
 }
 
 bool
@@ -144,6 +156,7 @@ listener_add_topic(struct listener *l, const struct topic *t)
     /* do nothing if already subscribed */
     TAILQ_FOREACH(node, l->idx, i_tqe) {
         if (node->obj == t) {
+            log_debug("listener %p already subscribed to topic %p", l, t);
             return false;
         }
     }
@@ -157,6 +170,8 @@ listener_add_topic(struct listener *l, const struct topic *t)
 
     TAILQ_INSERT_TAIL(l->idx, node, i_tqe);
     l->ntopic++;
+    log_verb("listener %p subscribed to topic %p, total subscription: %"PRIu32,
+            l, t, l->ntopic);
 
     return true;
 }
@@ -173,11 +188,15 @@ listener_del_topic(struct listener *l, const struct topic *t)
         }
     }
     if (node == NULL) {
+        log_debug("listener %p not subscribed to topic %p", l, t);
         return;
     }
 
     TAILQ_REMOVE(l->idx, node, i_tqe);
     l->ntopic--;
+    cc_free(node);
+    log_verb("listener %p unsubscribed from topic %p, total subscription: %"
+            PRIu32, l, t, l->ntopic);
 }
 
 void
diff --git a/src/storage/pubsub/listener.h b/src/storage/pubsub/listener.h
index e0273d229..5ea91b5ea 100644
--- a/src/storage/pubsub/listener.h
+++ b/src/storage/pubsub/listener.h
@@ -21,12 +21,8 @@ void listener_setup(uint32_t hash_power);
 void listener_teardown(void);
 
 struct listener *listener_get(const struct buf_sock *s);
-void listener_put(const struct listener *l);
+struct listener *listener_add(const struct buf_sock *s);
 void listener_delete(const struct buf_sock *s);
 
-struct listener *listener_create(struct buf_sock *s);
-void listener_destroy(struct listener **l);
-void listener_reset(struct listener *l);
-
 bool listener_add_topic(struct listener *l, const struct topic *t);
 void listener_del_topic(struct listener *l, const struct topic *t);
diff --git a/src/storage/pubsub/topic.c b/src/storage/pubsub/topic.c
index f6f4e1b4b..61609dd39 100644
--- a/src/storage/pubsub/topic.c
+++ b/src/storage/pubsub/topic.c
@@ -27,6 +27,64 @@ _get_bucket(const struct bstring *name)
     return &(ht->table[hval & HASHMASK(ht->hash_power)]);
 }
 
+static void
+_topic_reset(struct topic *t)
+{
+    t->name = null_bstring;
+    t->nsub = 0;
+    TAILQ_INIT(t->idx);
+}
+
+static struct topic *
+_topic_create(const struct bstring *name)
+{
+    struct topic *t;
+
+    t = cc_alloc(sizeof(struct topic));
+    if (t == NULL) {
+        return NULL;
+    }
+
+    t->idx = cc_alloc(sizeof(struct index_tqh));
+    if (t->idx == NULL) {
+        cc_free(t);
+        return NULL;
+    }
+
+    _topic_reset(t);
+    t->name.len = name->len;
+    t->name.data = cc_alloc(t->name.len);
+    if (t->name.data == NULL) {
+        cc_free(t->idx);
+        cc_free(t);
+        return NULL;
+    }
+    cc_memcpy(t->name.data, name->data, name->len);
+
+    return t;
+}
+
+static void
+_topic_destroy(struct topic **t)
+{
+    ASSERT(t != NULL && *t != NULL);
+
+    struct index_node *curr, *next;
+    struct index_tqh *idx = (*t)->idx;
+
+    /* delete all elements of the index */
+    TAILQ_FOREACH_SAFE(curr, idx, i_tqe, next) {
+        TAILQ_REMOVE(idx, curr, i_tqe);
+        cc_free(curr);
+    }
+    cc_free(idx);
+    cc_free((*t)->name.data);
+
+    cc_free(*t);
+    *t = NULL;
+}
+
+
 struct topic *
 topic_get(const struct bstring *name)
 {
@@ -35,26 +93,32 @@ topic_get(const struct bstring *name)
 
     bucket = _get_bucket(name);
     for (t = SLIST_FIRST(bucket); t != NULL; t = SLIST_NEXT(t, t_sle)) {
-        log_verb("current topic name (len %d) is: %.*s", t->name.len, t->name.len, t->name.data);
         if (bstring_compare(&t->name, name) == 0) {
             return t;
         }
     }
 
+    log_verb("topic not found name %.*s", t->name.len, t->name.data);
     return NULL;
 }
 
-void
-topic_put(const struct topic *t)
+struct topic *
+topic_add(const struct bstring *name)
 {
+    ASSERT(topic_get(name) == NULL);
+
     struct topic_slh *bucket;
+    struct topic *t = _topic_create(name);
 
-    ASSERT(topic_get(&t->name) == NULL);
+    log_verb("add topic %p for name %.*s", t, name->len, name->data);
 
     bucket = _get_bucket(&t->name);
-    SLIST_INSERT_HEAD(bucket, (struct topic *)t, t_sle);
+    SLIST_INSERT_HEAD(bucket, t, t_sle);
 
     ht->ntopic++;
+    log_verb("total topics: %"PRIu32, ht->ntopic);
+
+    return t;
 }
 
 void
@@ -81,65 +145,9 @@ topic_delete(const struct bstring *name)
         SLIST_REMOVE_AFTER(prev, t_sle);
     }
 
+    _topic_destroy(&t);
     --(ht->ntopic);
-}
-
-
-struct topic *
-topic_create(const struct bstring *name)
-{
-    struct topic *t;
-
-    t = cc_alloc(sizeof(struct topic));
-    if (t == NULL) {
-        return NULL;
-    }
-
-    t->idx = cc_alloc(sizeof(struct index_tqh));
-    if (t->idx == NULL) {
-        cc_free(t);
-        return NULL;
-    }
-
-    topic_reset(t);
-    t->name.len = name->len;
-    t->name.data = cc_alloc(t->name.len);
-    if (t->name.data == NULL) {
-        cc_free(t->idx);
-        cc_free(t);
-        return NULL;
-    }
-    cc_memcpy(t->name.data, name->data, name->len);
-
-    return t;
-}
-
-void
-topic_destroy(struct topic **t)
-{
-    ASSERT(t != NULL && *t != NULL);
-
-    struct index_node *curr, *next;
-    struct index_tqh *idx = (*t)->idx;
-
-    /* delete all elements of the index */
-    TAILQ_FOREACH_SAFE(curr, idx, i_tqe, next) {
-        TAILQ_REMOVE(idx, curr, i_tqe);
-        cc_free(curr);
-    }
-    cc_free(idx);
-    cc_free((*t)->name.data);
-
-    cc_free(*t);
-    *t = NULL;
-}
-
-void
-topic_reset(struct topic *t)
-{
-    t->name = null_bstring;
-    t->nsub = 0;
-    TAILQ_INIT(t->idx);
+    log_verb("total topics: %"PRIu32, ht->ntopic);
 }
 
 bool
@@ -152,18 +160,22 @@ topic_add_listener(struct topic *t, const struct listener *l)
     /* do nothing if already subscribed */
     TAILQ_FOREACH(node, t->idx, i_tqe) {
         if (node->obj == l) {
+            log_debug("topic %p already subscribed by listener %p", t, l);
             return false;
         }
     }
 
     node = cc_alloc(sizeof(struct index_node));
     if (node == NULL) {
+        log_error("cannot add listener: out of memory");
         return false;
     }
     node->obj = (struct listener *)l;
 
     TAILQ_INSERT_TAIL(t->idx, node, i_tqe);
     t->nsub++;
+    log_verb("topic %p subscribed by listener %p, total listeners: %"PRIu32,
+            t, l, t->nsub);
 
     return true;
 }
@@ -180,12 +192,16 @@ topic_del_listener(struct topic *t, const struct listener *l)
         }
     }
     if (node == NULL) {
+        log_debug("topic %p not subscribed by listener %p", t, l);
         return;
     }
 
     TAILQ_REMOVE(t->idx, node, i_tqe);
     t->nsub--;
     cc_free(node);
+    log_verb("topic %p subscribed by listener %p, total listeners: %"PRIu32,
+            t, l, t->nsub);
+
 }
 
 void
diff --git a/src/storage/pubsub/topic.h b/src/storage/pubsub/topic.h
index d8370fea7..b26a744a2 100644
--- a/src/storage/pubsub/topic.h
+++ b/src/storage/pubsub/topic.h
@@ -22,12 +22,8 @@ void topic_setup(uint32_t hash_power);
 void topic_teardown(void);
 
 struct topic *topic_get(const struct bstring *name);
-void topic_put(const struct topic *t);
+struct topic *topic_add(const struct bstring *name);
 void topic_delete(const struct bstring *name);
 
-struct topic *topic_create(const struct bstring *name);
-void topic_destroy(struct topic **t);
-void topic_reset(struct topic *t);
-
 bool topic_add_listener(struct topic *t, const struct listener *l);
 void topic_del_listener(struct topic *t, const struct listener *l);

From f7012c5a2927dd81a71d86bd10c65311a6c86866 Mon Sep 17 00:00:00 2001
From: Yao Yue <yao@twitter.com>
Date: Thu, 12 Apr 2018 18:19:43 -0700
Subject: [PATCH 9/9] fixing memory leak

---
 src/server/pubsub/data/process.c | 23 +++++++++++++++++++----
 src/server/pubsub/data/process.h |  2 ++
 2 files changed, 21 insertions(+), 4 deletions(-)

diff --git a/src/server/pubsub/data/process.c b/src/server/pubsub/data/process.c
index 1099387a8..5b8b6d1c6 100644
--- a/src/server/pubsub/data/process.c
+++ b/src/server/pubsub/data/process.c
@@ -253,7 +253,7 @@ pubsub_process_read(struct buf_sock *s)
         status = parse_req(req, s->rbuf);
         if (status == PARSE_EUNFIN) {
             buf_lshift(s->rbuf);
-            return 0;
+            goto done;
         }
         if (status != PARSE_OK) {
             /* parsing errors are all client errors, since we don't know
@@ -262,7 +262,9 @@ pubsub_process_read(struct buf_sock *s)
              * ends), we should close the connection
              */
             log_warn("illegal request received, status: %d", status);
-            return -1;
+            INCR(process_metrics, process_ex);
+            INCR(process_metrics, process_client_ex);
+            goto error;
         }
 
         /* stage 2: processing- check for quit, allocate response(s), process */
@@ -270,24 +272,37 @@ pubsub_process_read(struct buf_sock *s)
         /* quit is special, no processing/resposne expected */
         if (req->type == REQ_QUIT) {
             log_info("peer called quit");
-            return -1;
+            goto error;
         }
 
         /* actual processing */
         process_request_sock(rsp, req, s);
 
         /* stage 3: write response(s) if necessary */
-        compose_rsp(&s->wbuf, rsp);
+        status = compose_rsp(&s->wbuf, rsp);
+        if (status != PARSE_OK) {
+            log_error("composing rsp erred");
+            INCR(process_metrics, process_ex);
+            INCR(process_metrics, process_server_ex);
+            goto error;
+        }
 
         /* noreply means no need to write to buffers */
 
         /* logging, clean-up */
     }
 
+done:
     request_return(&req);
     response_return(&rsp);
 
     return 0;
+
+error:
+    request_return(&req);
+    response_return(&rsp);
+
+    return -1;
 }
 
 
diff --git a/src/server/pubsub/data/process.h b/src/server/pubsub/data/process.h
index c7d49912f..63f06cc52 100644
--- a/src/server/pubsub/data/process.h
+++ b/src/server/pubsub/data/process.h
@@ -11,6 +11,8 @@
 #define PROCESS_METRIC(ACTION)                                          \
     ACTION( process_req,       METRIC_COUNTER, "# requests processed"  )\
     ACTION( process_ex,        METRIC_COUNTER, "# processing error"    )\
+    ACTION( process_client_ex, METRIC_COUNTER, "# client error"        )\
+    ACTION( process_server_ex, METRIC_COUNTER, "# server error"        )\
     ACTION( publish,           METRIC_COUNTER, "# publish requests"    )\
     ACTION( subscribe,         METRIC_COUNTER, "# subscribe requests"  )\
     ACTION( unsubscribe,       METRIC_COUNTER, "# unsubscribe requests")