Threaded vote entry using redis PoC

threaded-entry-redis-PoC.py

from threading import Thread, get_ident
import random
import redis
from time import sleep

from pathlib import Path


redisServer = redis.Redis(host='localhost', port=6379, db=5, decode_responses=True)

#TTL = 60*10
TTL = 60
SLEEPTIME_MIN=10
SLEEPTIME_MAX=30
MAX_THREADS=4

TODO_PREFIX = 'todo'
BUSY_PREFIX = 'busy'


def empty_redis():
    for key in redisServer.scan_iter('{0}:*'.format(BUSY_PREFIX)):
        redisServer.delete(key)

    for key in redisServer.scan_iter('{0}:*'.format(TODO_PREFIX)):
        redisServer.delete(key)


def get_files():
    files = []

    for path in Path('/home/richard/tmp/google_signin_buttons').rglob('*'):
        files.append(str(path))

    return files


def fill_redis(files):
    for filename in files:
        redisServer.set(get_todo_key(filename), filename)


def get_busy_key(filename):
    return '{0}:{1}'.format(BUSY_PREFIX, filename)


def get_todo_key(filename):
    return '{0}:{1}'.format(TODO_PREFIX, filename)


def get_new_item():
    for key in redisServer.keys('{0}:*'.format(TODO_PREFIX)):
        filename = redisServer.get(key)
        busy_key = get_busy_key(filename)
        busy_item = redisServer.get(busy_key)
        if busy_item is None:
            print('not busy, returning file')
            redisServer.set(busy_key, filename, ex=TTL)

            return filename
        else:
            print('filename busy, continue')


def process_file(filename):
    id = get_ident()
    print('{0}: processing {1}'.format(id, filename))
    sleeptime = random.randint(SLEEPTIME_MIN, SLEEPTIME_MAX)
    print('{0}: sleeping {1} minutes'.format(id, sleeptime/60))
    sleep(sleeptime)
    print('{0}: done, removing busy_key and todo key'.format(id))
    redisServer.delete(get_busy_key(filename))
    redisServer.delete(get_todo_key(filename))


threads = []
empty_redis()
files = get_files()
fill_redis(files)

count = len(redisServer.keys('{0}:*'.format(TODO_PREFIX)))
while count > 0:
    for thread in threads:
        if not thread.is_alive():
            threads.remove(thread)

    num_threads = len(threads)

    print('waiting, count={0}, num_threads={1}'.format(count, num_threads))
    sleep(1)

    if num_threads < MAX_THREADS:
        filename = get_new_item()
        print('creating thread')
        log_thread = Thread(args=(filename,), target=process_file)
        log_thread.start()
        threads.append(log_thread)

    count = len(redisServer.keys('{0}:*'.format(TODO_PREFIX)))