monolith/modules/chankeep.py

import main
from utils.logging.log import *
from utils.logging.debug import *
from copy import deepcopy
from math import ceil
from modules.provision import provisionMultipleRelays
from twisted.internet.threads import deferToThread
from numpy import array_split

def allRelaysActive(net):
    relayNum = len(main.network[net].relays.keys())
    existNum = 0
    for i in main.network[net].relays.keys():
        name = net+str(i)
        if name in main.IRCPool.keys():
            existNum += 1
    if existNum == relayNum:
        return True
    return False

def populateChans(net, clist, relay):
    divided = array_split(clist, relay)
    for i in range(0, len(divided)):
        if net in main.TempChan.keys():
            main.TempChan[net][i] = divided[i]
        else:
            main.TempChan[net] = {i: divided[i]}

def notifyJoin(net):
    for i in main.network[net].relays.keys():
        name = net+str(i)
        if name in main.IRCPool.keys():
            main.IRCPool[name].checkChannels()

def keepChannels(net, listinfo, mean, sigrelay, relay):
    #print("list", listinfo)
    #print("sigrelay", sigrelay)
    #print("cur", len(main.network[net].relays.keys()))
    if relay <= main.config["ChanKeep"]["SigSwitch"]: # we can cover all of the channels
        coverAll = True
    elif relay > main.config["ChanKeep"]["SigSwitch"]: # we cannot cover all of the channels
        coverAll = False
        if not sigrelay <= main.config["ChanKeep"]["MaxRelay"]:
            error("Network %s is too big to cover: %i relays required" % (net, sigrelay))
            return
    if coverAll:
        needed = relay-len(main.network[net].relays.keys())
        flist = [i[0] for i in listinfo]
        populateChans(net, flist, relay)
    else:
        needed = sigrelay-len(main.network[net].relays.keys())
        siglist = [i[0] for i in listinfo if int(i[1]) > mean]
        populateChans(net, siglist, sigrelay)
    notifyJoin(net)
    if needed > 0:
        provisionMultipleRelays(net, needed)

    #print("coverall", coverAll)
    #print("needed", needed)

def purgeRecords(net):
    base = "list.%s" % net
    p = main.g.pipeline()
    existingChans = main.g.smembers("list."+net)
    for i in existingChans:
        p.delete(base+"."+i.decode("utf-8"))
    p.execute()

def _nukeNetwork(net):
    purgeRecords(net)
    p = main.g.pipeline()
    p.delete("analytics.list."+net)
    p.delete("list."+net)
    p.execute()

def nukeNetwork(net):
    deferToThread(_nukeNetwork, net)

def _initialList(net, num, listinfo, chanlimit):
    #listinfo = sorted(listinfo, key=lambda x: xdd[0])
    listlength = len(listinfo)
    cumul = 0
    try:
        cumul += sum(int(i[1]) for i in listinfo)
    except TypeError:
        warn("Bad LIST data received from %s - %i" % (net, num))
        return
    mean = round(cumul/listlength, 2)
    siglength = 0
    insiglength = 0
    sigcumul = 0
    insigcumul = 0
    for i in listinfo:
        if int(i[1]) > mean:
            siglength += 1
            sigcumul += int(i[1])
        elif int(i[1]) < mean:
            insiglength += 1
            insigcumul += int(i[1])

    if not net in main.network.keys():
        warn("Cannot write list info - no network entry for %s" % net)
        return
    sigrelay = ceil(siglength/chanlimit)
    relay = ceil(listlength/chanlimit)
    netbase = "list.%s" % net
    abase = "analytics.list.%s" % net
    p = main.g.pipeline()
    p.hset(abase, "mean", mean)
    p.hset(abase, "total", listlength)
    p.hset(abase, "sigtotal", siglength)
    p.hset(abase, "insigtotal", insiglength)
    p.hset(abase, "sigperc", round(siglength/listlength*100, 2))
    p.hset(abase, "insigperc", round(insiglength/listlength*100, 2))
    p.hset(abase, "cumul", cumul)
    p.hset(abase, "sigcumul", sigcumul)
    p.hset(abase, "insigcumul", insigcumul)
    p.hset(abase, "relay", relay)
    p.hset(abase, "sigrelay", sigrelay)
    p.hset(abase, "insigrelay", ceil(insiglength/chanlimit))
    # Purge existing records before writing
    purgeRecords(net)
    for i in listinfo:
        p.rpush(netbase+"."+i[0], i[1])
        p.rpush(netbase+"."+i[0], i[2])
        p.sadd(netbase, i[0])

    p.execute()
    debug("List parsing completed on %s" % net)
    keepChannels(net, listinfo, mean, sigrelay, relay)

def initialList(net, num, listinfo, chanlimit):
    deferToThread(_initialList, net, num, deepcopy(listinfo), chanlimit)