foghorn.py

#caution, only use this script if you have no important data in your redis db with index 9

#idea: - create the most naive and simple foghorn implementation possible with redis.
#      - only create the server, not the clients who actually deliver the messages out, those can be scaled vertically
#      - only the messaging core, no http stuff
#
#reason: get a feeling for the problems and do a some benchmark
#outcome: while you can make redis the bottleneck with multiple instances of this type of 
#   Foghorn sever, it scales not that well.
# - the set intersection to get users from a group is O(N). a btree + mergesort would be faster here (sql...)
# - with this approach there are many more messages in the queue than necessairy. see comment in the publish method.
# - pickling
#
# why not use redis' pub/sub mechanism: it will probably scale even worse, as you would need a redis client for each user.
# also this approach will not account for lost msgs due to client restarts/code refresh

from redis import Redis
import pickle

r = Redis(decode_responses=True, db=9) #TODO: needs threadpool and/or singleton

class Foghorn():
    
    def __init__(self):
        self.redis = r #TODO: threadpool
    
    #publishes a message to all subscribers of that topic
    #the reason to create a message for each subscriber is so that the
    #sender can get feedback on how many msgs his call resulted in and 
    #additionally minimize the time window on which subscribers can change for the channel.
    def publish(self, sender, topic, txt):
        r = self.redis
        #get all groups for topic
        memberset = "topics:%s:groups" % topic
        group_ids = r.smembers(memberset)
        #get all users for groups
        groups = map(lambda y: "groups:%s:members" % y ,group_ids)
        #check if user can post into that topic
        user_ids = r.sunion(*groups)
        
        #enqueue messags to users
        for user_id in user_ids:
            Message(sender, user_id, topic, txt ).save()
            
    def add_user(self, username):
        self.add_user_to_group(username, username) #create a new group with the new user as sole member
        
    def add_user_to_group(self, username, group):
        self.redis.sadd("groups:%s:members" % group, username)
    
    def add_group_to_topic(self,group,topic):
        self.redis.sadd("topics:%s:groups" % topic, group)
    
    

class Message():
    
    def __init__(self, sender, receiver, topic, txt):
        self.sender = sender
        self.txt = txt
        self.topic = topic
        #TODO: lookup messaging method according to users configuration for this topic
        self.type = "debug"
        
    def save(self): 
        r.rpush("msgs:%s" % self.type, pickle.dumps(self)) #TODO use msgpack or capnproto
        
        
        
    
from time import time
import random
random.seed()   
max_users_per_group = 10
max_users_per_topic = 50
messages_to_add = 100
def bench():
    Redis(decode_responses=True, db=9).flushdb()
    users = ["user%s" %i for i in range(50)]
    groups = ["randomgroup%s" %i for i in range(50)]
    topics = ["topic%s" %i for i in range(20)]
    f = Foghorn()
    #add all users
    print("add %i users" % len(users))
    for user in users:
        f.add_user(f)
    #add random amount of users into each groups
    print("add random amout of users into %i groups" % len(groups)) 
    for grp in groups:
        for x in range(1, random.randrange(1, max_users_per_group)):
            f.add_user_to_group(random.choice(users), grp)
    
    #subscribe random amount of groups to each topic
    print("subscribe random amount of groups to %i topics" % len(topics))
    for topic in topics:
        for x in range(1, random.randrange(1, max_users_per_topic)):
            f.add_group_to_topic(random.choice(groups), topic)
        #add at least one
    #now send some messsages
    print("party hard")
    start = time()
    for x in range(messages_to_add):
        sender = random.choice(users)
        topic = random.choice(topics)
        f.publish(sender, topic, "msg %s" % x)
  
    end = time()
    return  messages_to_add /  (end - start)