LiF_StreamingBandit_AsyncIO_Server_Benchmark.py

# -*- coding: utf-8 -*-

#from IPython import get_ipython
#get_ipython().magic('reset -sf')


import sys,random
import ujson as json
import asyncio,aiohttp,pymongo
import numpy as np
import matplotlib.pyplot as plt

MONGO_IP = "127.0.0.1"
MONGO_PORT = 27017
SB_BASE_URL = "http://127.0.0.1:8080"

print_progress = 1
exp_key = "95113906f"
exp_id = 1
question_nr = 12345

parallel_processes = 4
repeats_per_process = 500

max_time_set_reward = 5.0
sd_participants = 3
drop_probability = 0.8

start_at_t = 0

def main():
    req_version = (3,5)
    cur_version = sys.version_info
    if cur_version >= req_version:

        # Necessary to be able to run more than once in same console in Win
        if sys.platform == 'win32':
            loop = asyncio.ProactorEventLoop()
            asyncio.set_event_loop(loop)
        else :
            loop = asyncio.SelectorEventLoop()
            asyncio.set_event_loop(loop)

        future = asyncio.Future()
        asyncio.ensure_future(do_async(future))
        loop.run_until_complete(future)

        loop.stop()
        loop.close()

        do_chart()

    else:
       print("This script makes use of asyncio features that where introduced in 3.5")
       print("Consider upgrading your Python interpreter to 3.5 or higher.")
       sys.exit(0)

async def do_async(future):
   for i in range(parallel_processes):
        asyncio.ensure_future(get_action(future))

async def get_action(future):
    for i in range(repeats_per_process):
        request =  SB_BASE_URL + "/" + str(exp_id) +"/getAction.json?key="+exp_key
        request += "&context={\"question\":"+str(question_nr)+"}"
        response = await aiohttp.request('GET', request,encoding='utf-8',)
        #await asyncio.sleep(0.0001)
        body = await response.read_and_close(decode=False)
        response.close()
        obj = json.loads(body.decode('utf-8'))
        t = obj["action"]["t"]
        x = obj["action"]["x"]
        global start_at_t
        if (start_at_t==0):
            start_at_t = t
        if print_progress:
            print("#####"+str(t))
        asyncio.ensure_future(set_reward(t,x,future))

async def set_reward(t,x,future):
    if np.random.binomial(1, drop_probability, 1)==1 or True:
           y =  -1*pow((x-5),2) + np.random.normal(0,sd_participants,1)
           request =  SB_BASE_URL + "/" + str(exp_id) + "/setReward.json"
           request += "?key="+exp_key
           request += "&context={\"question\":"+str(question_nr)+"}"
           request += "&action={\"x\":" + str(float(x))
           request += ",\"t\":" + str(float(t)) + "}"
           request += "&reward=" + str(float(y))
           await asyncio.sleep(random.uniform(0.0, max_time_set_reward))
           response = await aiohttp.request('GET', request ,encoding='utf-8',)
           response.close()
           if print_progress:
               print("$$$$$$$$$$"+str(t))
           if t>=repeats_per_process*parallel_processes+start_at_t-1:
               await asyncio.sleep(max_time_set_reward+1.0)
               future.set_result(future)

def do_chart():

    client = pymongo.MongoClient(MONGO_IP, MONGO_PORT)
    db = client.logs
    cursor = db.logs.find({"type": "setreward","q":question_nr}).sort([("t", pymongo.ASCENDING)])
    result_list = list(cursor)
    client.close();

    t = [ts['t'] for ts in result_list]
    x0 = [xs['x0'] for xs in result_list]

    fig = plt.figure(figsize=(30,7))
    ax = fig.add_subplot(1,1,1)
    major_ticks = np.arange(0, len(t), 100)
    minor_ticks = np.arange(0, len(t), 50)
    ax.tick_params(which = 'both', direction = 'out')
    ax.set_xticks(major_ticks)
    ax.set_xticks(minor_ticks, minor=True)
    ax.grid(which='both')

    plt.plot(t,x0)
    plt.show()

if __name__ == '__main__':
    main()