<jupyter_start><jupyter_text>Memcached Data LoadingThis notebook loads memcached data for subsequent analysis<jupyter_code>%matplotlib notebook
import dpkt
import os
import struct
import numpy as np
import pandas as pd
from collections import defaultdict
from pprint import pprint
from collections import namedtuple<jupyter_output><empty_output><jupyter_text>Memoizer stores data based on input arguments, making it easier to run repeated analysis.Check out here with```git clone https://github.com/isaac-ped/Memoizer```(or don't and the wrapper `@MEMO` will do nothing)<jupyter_code>memo = lambda x: x
try:
    from Memoizer import memoize_to_folder
    memo = memoize_to_folder('mcd_memoization')
except Exception as e:
    print("Warning: could not load memoizer")<jupyter_output><empty_output><jupyter_text>`ray` helps with task paralellization so loading data from pcaps takes much less timeIf `ray` cannot be loaded, `HAS_RAY` is set to `False` and analysis will proceed serially<jupyter_code>HAS_RAY=False
try:
    import ray
    ray.shutdown()
    ray.init(num_cpus=12)
    HAS_RAY=True
except Exception as e:
    print("Warning: Could not load ray: %s" % e)<jupyter_output>2019-09-13 13:49:42,401	WARNING worker.py:1373 -- WARNING: Not updating worker name since `setproctitle` is not installed. Install this with `pip install setproctitle` (or ray[debug]) to enable monitoring of worker processes.
2019-09-13 13:49:42,403	INFO node.py:498 -- Process STDOUT and STDERR is being redirected to /tmp/ray/session_2019-09-13_13-49-42_403143_959/logs.
2019-09-13 13:49:42,519	INFO services.py:409 -- Waiting for redis server at 127.0.0.1:20159 to respond...
2019-09-13 13:49:42,641	INFO services.py:409 -- Waiting for redis server at 127.0.0.1:39078 to respond...
2019-09-13 13:49:42,647	INFO services.py:809 -- Starting Redis shard with 5.89 GB max memory.
2019-09-13 13:49:42,678	INFO node.py:512 -- Process STDOUT and STDERR is being redirected to /tmp/ray/session_2019-09-13_13-49-42_403143_959/logs.
2019-09-13 13:49:42,681	INFO services.py:1475 -- Starting the Plasma object store with 8.84 GB memory using /dev/shm.<jupyter_text>Pcap readingRead individual pcap files into a dataframe with the packet type and the time at which it went through the tofino<jupyter_code>KEYLEN=8

# PktTime defines the columns of the output of read_memcached_pcap
PktTime = namedtuple('pkt_time', ( 'type', 'key', 'time', 'hit'))

@memo
def read_memcached_pcap(out_dir, dst_mac_is_ts = True, compare_counters = True, force_ipv4=True):
    ''' Reads the moongen pcap file in the provided output directory.
    Assumes the pcap file is located in <out_dir>/moongen/moongen.pcap
    Returns dataframe with columns from the named tuple PktTime'''
    if compare_counters:
        count_file = os.path.join(out_dir, 'counters_0.out')
        for line in open(count_file):
            if line.startswith('5/0'):
                n_pkts_str = line.split()[-1]
                n_start_pkts = int(n_pkts_str[:-1])
        count_file = os.path.join(out_dir, 'counters_1.out')
        for line in open(count_file):
            if line.startswith('5/0'):
                n_pkts_str = line.split()[-1]
                n_end_pkts = int(n_pkts_str[:-1])   
        n_expected_pkts = (n_end_pkts - n_start_pkts)
        
        print("Expecting {} packets".format(n_expected_pkts))
                
    pcf = os.path.join(out_dir, 'moongen', 'moongen.pcap')
    
    pkt_times = []

    f = open(pcf, 'rb')
    print("Reading {}".format(pcf))
    pcap=dpkt.pcap.Reader(f)
    total_size = 0
    start_ts = None
    for i, (ts, buf) in enumerate(pcap):
        if i % int(n_expected_pkts / 10) == 0:
            print(i)
        try:
            eth=dpkt.ethernet.Ethernet(buf)

            if dst_mac_is_ts:
                x3, x2, x1 = struct.unpack("!HHH", eth.dst)
                ts = float((x3 << 32) | (x2 << 16) | x1) * 1e-9
                if start_ts is None:
                    start_ts = ts
                ts -= start_ts
            
            if force_ipv4:
                ip = eth.data
                if isinstance(ip, bytes):
                    try:
                        ip = dpkt.ip.IP(eth.data)
                    except (dpkt.UnpackError) as e:
                        print(e)
                        pkt_times.append(PktTime('other (error)', '', ts, -1))
                        continue
                elif not isinstance(ip, dpkt.ip.IP):
                    pkt_times.append(PktTime('other', '', ts, -1))
                    continue
            else:
                ip=eth.data
                if not isinstance(ip, dpkt.ip.IP):
                    pkt_times.append(PktTime('other', '', ts, -1))
                    continue
            udp = ip.data
            spkt = udp.data
            
        except Exception as e:
            print("Exception parsing packet {} : {}".format(buf, e))
            raise

        if type(spkt) == dpkt.icmp.ICMP.Unreach:
            pkt_times.append(PktTime('unreach', '', ts, -1))
            continue

        get_start = spkt.find(b'get ')
        val_start = spkt.find(b'VALUE ')
        hit = -1
        if get_start >= 0:
            key = spkt[:2] + spkt[get_start+4:get_start+4+KEYLEN]
            pkt_type = 'get'
        elif val_start >= 0:
            key = spkt[:2] + spkt[val_start+6:val_start+6+KEYLEN]
            pkt_type = 'value'
            hit = ip.id
        elif b'set ' in spkt:
            key = spkt[:2]
            pkt_type = 'set'
        elif b'STORED' in spkt:
            key = spkt[:2]
            pkt_type = 'stored'
        elif b'END' in spkt:
            key = spkt[:2]
            pkt_type = 'miss'
        else:
            key = ''
            pkt_type = 'other'
            print('Unknown packet load: {}'.format(buf))
        pkt_times.append(PktTime(pkt_type, key, ts, hit))
    
    print("Finished reading pcap: ")
    df = pd.DataFrame(pkt_times)
    print(df.groupby('type').size())
    
    if len(df) < n_expected_pkts - 1:
        print("******************\n"*5 + \
              "GOT {} PKTS, EXPECTED {}\n".format(len(df), n_expected_pkts) +\
              "******************\n"*5
             )        
        
    return df.sort_values('time').reset_index(drop=True)<jupyter_output><empty_output><jupyter_text>Dataframe analysisMatch requests with subsequent responses.Create two dataframes:* Latency of all successful requests* Percentage of requests that were successful over time The following functions help with matching requests with corresponding responses, made trickier by the fact that keys are repeated over time.An assumption is made that a request key is not repeated until its response is gathered.As request keys are a combination of memcached KEY and the 2-byte udp ID field (see memcached documentation for details on udp Id), and UDP IDs are given in order, there is a guarantee that requests will not repeat for 65,536 packets.At 10Gbps, 65,536 packets are sent every ~$13 ms$$65536~pkts * (10~Gbps~/~( 8~bits~/~Byte )~/~(256~bytes~/~packet)) = 13.4ms$<jupyter_code>65536 / (10e9 / 8 / 256) # packets per second<jupyter_output><empty_output><jupyter_text>Since all latencies are well under 5 ms, this shouldn't be a concern<jupyter_code>def repetition_count(keys):
    '''Produces a column with the repetition number of a set of keys.
    (i.e., if keys is [1, 1, 3, 0, 1, 0], will output [0, 1, 0, 0, 2, 1])
    '''
    dup_nums = np.zeros(len(keys), dtype=int)
    keys = keys.astype(object)
    for i in range(1, 100):
        dups = keys.duplicated()
        if dups.any():
            dup_nums[dups] += 1
            keys[dups] = keys[dups] + b'_dup'
        else:
            break
    else:
        print("WARNING: %d duplicates still remain" % (np.count_nonzero(keys.duplicated())))
    return dup_nums

def match_responses(reqs, resps):
    '''Produces a single dataframe with only successful requests and corresponding latencies'''
    reqs = reqs.assign(dup = repetition_count(reqs.key))
    resps = resps.assign(dup = repetition_count(resps.key))
    
    merged = pd.merge(reqs, resps, on='key', how='right', suffixes=('_req', '_resp'))
    merged['latency'] = merged['time_resp'] - merged['time_req']
    
    # Sort by latency, so that in the groupby(), the lowest latency will be first()
    merged = merged[merged['latency'] > 0].sort_values('latency')
    # NB: Assumes that the lowest latency is the correct latency
    merged = merged.groupby(['key', 'dup_resp']).first().reset_index()
    
    merged = merged.sort_values('time_req').reset_index(drop=True)
    merged = merged.rename(columns={'type_req': 'type', 'time_req': 'time', 'hit_resp': 'hit'})
    return merged[['key', 'type', 'time', 'latency', 'hit']]<jupyter_output><empty_output><jupyter_text>These functions produce condensed dataframes with latencies or successes over time<jupyter_code># The success dataframe is split into intervals containing this many packets 
# per request-type (set/get) per experiment
PKTS_PER_INTERVAL = 500

def request_success(reqs, resps, matched=None, pkts_per_interval = PKTS_PER_INTERVAL):
    '''Returns a dataframe denoting percentage success over time'''
    reqs = reqs.copy()
    resps = resps.copy()
    
    # Uses match_responses to ensure requests are matched with the correct responses
    if matched is None:
        matched = match_responses(reqs, resps)
    
    interval = (reqs.time.max() - reqs.time.min()) / (len(reqs) / pkts_per_interval)
    matched['interval'] = (matched.time / interval).apply(np.floor)
    reqs['interval'] = (reqs.time / interval).apply(np.floor)
    
    ratio = (matched.groupby('interval').size() / reqs.groupby('interval').size()).rename('success')
    ratio.index *= interval
    ratio.index = ratio.index.rename('time')
    return pd.DataFrame(ratio)
    
def experiment_success(pkts):
    '''For an experiment's packets, returns get/set success over time in a single dataframe'''
    get_success = request_success(pkts[pkts.type == 'get'], pkts[pkts.type == 'value'])
    get_success['type'] = 'get'
    set_success = request_success(pkts[pkts.type == 'set'], pkts[pkts.type == 'stored'])
    set_success['type'] = 'set'
    succ = pd.concat([get_success, set_success], axis=0).reset_index()
    return succ.rename(columns={'success': 'value'})

def experiment_latencies(pkts):
    '''For an experiment's packets, returns get/set latencies per request in a single dataframe'''
    get_lat = match_responses(pkts[pkts.type == 'set'].copy(), pkts[pkts.type == 'stored'].copy())
    set_lat = match_responses(pkts[pkts.type == 'get'].copy(), pkts[pkts.type == 'value'].copy())
    lat = pd.concat([get_lat, set_lat], axis=0).sort_values('time').reset_index(drop=True)
    return lat.rename(columns={'latency': 'value'})

def experiment_stats(pkts):
    '''Returns tuple of dataframes (latency, success) for an experiment'''
    sets = pkts[pkts.type == 'set']
    stored = pkts[pkts.type == 'stored']
    get = pkts[pkts.type == 'get']
    value = pkts[pkts.type == 'value']
    
    set_lat = match_responses(sets, stored)
    get_lat = match_responses(get, value)
    
    print("Got latencies:\n\tSET: %d\n\tGET: %d" % (len(set_lat), len(get_lat)))
    
    lat = pd.concat([get_lat, set_lat], axis=0).sort_values('time').reset_index(drop=True)
    
    get_succ = request_success(get, value, get_lat)
    get_succ['type'] = 'get'
    
    set_succ = request_success(sets, stored, set_lat)
    set_succ['type'] = 'set'
    
    succ = pd.concat([get_succ, set_succ], axis=0).reset_index()
    
    return lat.rename(columns={'latency': 'value'}), succ.rename(columns={'success': 'value'})<jupyter_output><empty_output><jupyter_text>AggregationThese functions help with aggregation across experiments, and with speeding up loading data<jupyter_code>@memo
def load_experiment_stats(base, experiment, label):
    ''' A memoizable function which loads and processes experiment data in a given location'''
    pkts = read_memcached_pcap(os.path.join(base, experiment, '%s_%s' % (experiment, label)))
    return experiment_stats(pkts)

# Can't "ray" a function decorated with `memo` because of **kwargs,
# so we must create a wrapper function
if HAS_RAY:
    @ray.remote
    def load_experiment_stats_parallel(*args):
        return load_experiment_stats(*args)

def aggregate_experiment_stats(base, experiments, labels, parallel = HAS_RAY):
    '''Loads and concatenates all stats for a list of experiments and labels.'''
    all_stats = {}
    for lab in labels:
        for exp in experiments:
            print("Aggregating %s:%s" % (exp, lab))
            if parallel:
                all_stats[(lab, exp)] = load_experiment_stats_parallel.remote(base, exp, lab)
            else:
                all_stats[(lab, exp)] = load_experiment_stats(base, exp, lab)
                
    if parallel:
        for k in all_stats.keys():
            all_stats[k] = ray.get(all_stats[k])
            
    all_latencies = []
    all_successes = []
    for (lab, exp), (lat, suc) in all_stats.items():
        lat = lat.assign(experiment=exp, exp_label=lab)
        suc = suc.assign(experiment=exp, exp_label=lab)
        all_latencies.append(lat)
        all_successes.append(suc)
        
    return (pd.concat(all_latencies, ignore_index=True),
            pd.concat(all_successes, ignore_index=True))<jupyter_output><empty_output><jupyter_text>Default data locationsBy default, experiments consist of these sets of data<jupyter_code>FPGA_BASE_DIR = '../e2e_output_500k_fpga_hc_14/'
TOFINO_BASE_DIR = '../e2e_output_500k_tofino_hc_14/'
ARISTA_BASE_DIR = '../e2e_output_500k_arista_no_passthrough_2/'
CPU_BASE_DIR = '../e2e_output_500k_cpu_hc_11/'


EXPERIMENTS = ['baseline', 'hc', 'drop', 'fec', 'kv']

DEV_BASE_DIRS = dict(
    fpga = FPGA_BASE_DIR,
    tofino = TOFINO_BASE_DIR,
    arista = ARISTA_BASE_DIR,
    cpu = CPU_BASE_DIR,
)

DEV_EXP_LABELS = dict(
    tofino = range(1, 6),
    fpga = range(1, 6),
    arista = range(1, 6),
    cpu = range(1, 6)
)<jupyter_output><empty_output><jupyter_text>Loading dataLoads all of the data for a given device and saves it in a pickle file easier loading and analysis later<jupyter_code>import pickle

def save_device_data(dev, experiments=EXPERIMENTS):
    save_file = '%s_mcd.pickle' % dev
    if os.path.exists(save_file):
        print("Loading from save file %s" % save_file)
        data = pickle.load(open(save_file, 'rb'))
        print("Loaded")
        return data
    data = aggregate_experiment_stats(DEV_BASE_DIRS[dev], experiments, DEV_EXP_LABELS[dev])
    print("Saving to file %s" % save_file)
    pickle.dump(data, open(save_file, 'wb'))
    return data
    
fpga_data = save_device_data('fpga')
tofino_data = save_device_data('tofino')
arista_data = save_device_data('arista')
cpu_data = save_device_data('cpu', ['baseline', 'hc'])<jupyter_output>Aggregating baseline:1
Aggregating hc:1
Aggregating baseline:2
Aggregating hc:2
Aggregating baseline:3
Aggregating hc:3
Aggregating baseline:4
Aggregating hc:4
Aggregating baseline:5
Aggregating hc:5
[2m[36m(pid=1088)[0m Loading from mcd_memoization/load_experiment_stats__base-e2e_output_500k_cpu_hc_11__experiment-baseline__label-2.pickle
[2m[36m(pid=1091)[0m Loading from mcd_memoization/load_experiment_stats__base-e2e_output_500k_cpu_hc_11__experiment-hc__label-5.pickle
[2m[36m(pid=1096)[0m Loading from mcd_memoization/load_experiment_stats__base-e2e_output_500k_cpu_hc_11__experiment-baseline__label-5.pickle
[2m[36m(pid=1089)[0m Loading from mcd_memoization/load_experiment_stats__base-e2e_output_500k_cpu_hc_11__experiment-baseline__label-4.pickle
[2m[36m(pid=1092)[0m Loading from mcd_memoization/load_experiment_stats__base-e2e_output_500k_cpu_hc_11__experiment-hc__label-1.pickle
[2m[36m(pid=1099)[0m Loading from mcd_memoization/load_experiment_stats__base-e2e_output_[...]<jupyter_text>Sample data plottingThe following cells contain sample plots of data to verify that experiments ran correctly<jupyter_code># plot some sample stats
import matplotlib.pyplot as plt

def plot_sample_stat(df, axs, column='value', exp_label=None, experiments=EXPERIMENTS, sample_size=None, roll_fn=None, color=None):
    if exp_label is None:
        exp_label = df.iloc[0].exp_label
    df_label = df[df.exp_label == exp_label]
    for ax, experiment in zip(axs, experiments):
        plt.sca(ax)
        df_exp = df_label[df_label.experiment == experiment]
        for pkt_type in ('get', 'set',):
            df_sample = df_exp[df_exp.type == pkt_type]

            if sample_size is not None:
                df_sample = df_sample.sample(sample_size)
  
            if roll_fn is None:
                handle = plt.plot(df_sample.time, df_sample[column], '.',  color=color, markersize=1, label=pkt_type)
            else:
                df_sample = df_sample.sort_values('time')
                df_sample['tdelta'] = pd.to_datetime(df_sample.time * 1e9)
                if roll_fn == 'median':
                    df_roll = df_sample.rolling(pd.Timedelta(1, 's'), on='tdelta')[['time', column]].median()
                elif roll_fn == 'mean':
                    df_roll = df_sample.rolling(pd.Timedelta(1, 's'), on='tdelta')[['time', column]].mean()
                handle = plt.plot(df_roll.time, df_roll[column], '-', color=color, label=pkt_type)
                
    return handle
         
def plot_group_stat(df, axs, column='value', start_time = 0, stop_time = 100, experiments=EXPERIMENTS):
    df = df[(df.time > start_time) & (df.time < stop_time)]
    for ax, experiment in zip(axs, experiments):
        plt.sca(ax)
        df_exp = df[df.experiment == experiment]
        for i, pkt_type in enumerate(['get', 'set']):
            df_type = df_exp[df_exp.type == pkt_type]
            grouped = df_type.groupby('exp_label')[column]
            latencies = grouped.median()
            lows, highs = grouped.quantile(.01), grouped.quantile(.99)
            plt.errorbar(latencies.index + i/3, latencies, yerr=[latencies-lows, highs-latencies], 
                         fmt='o', label=pkt_type, fillstyle='none')
        plt.title(experiment)
experiments = ['baseline', 'hc']
fig, axs = plt.subplots(2, 5, squeeze=False, sharey='row', sharex=True)
f = plot_sample_stat(fpga_data[0], axs[0,:], color='r')
t = plot_sample_stat(tofino_data[0], axs[0,:], color='g')
a = plot_sample_stat(arista_data[0], axs[0,:], color='b')

f = plot_sample_stat(fpga_data[0], axs[1,:], color='r', roll_fn='mean')
t = plot_sample_stat(tofino_data[0], axs[1,:], color='g', roll_fn='mean')
a = plot_sample_stat(arista_data[0], axs[1,:], color='b', roll_fn='mean')

# plt.legend(f+t+a, ['fpga', 'tofino', 'arista'], loc='upper right')
plt.legend(f+t+a, ['fpga', 'tofino', 'arista'], loc='upper right')

for exp, ax in zip(EXPERIMENTS, axs[0,:]):
    ax.set_title(exp)

axs[0][0].set_ylabel("Latency (s)")
axs[1][0].set_ylabel("1s rolling mean\nLatency (s)")
axs[1][0].set_yscale('log')
plt.tight_layout()
col = 'latency'
fig, axs = plt.subplots(2, len(EXPERIMENTS), squeeze=False, sharey='row', sharex=True)

f = plot_sample_stat(fpga_data[0], axs[1,:], col, color='r', sample_size=1000)
a = plot_sample_stat(arista_data[0], axs[1,:], col,  color='b', sample_size=1000)

f = plot_sample_stat(fpga_data[0], axs[0,:], col, color='r', sample_size=None)
a = plot_sample_stat(arista_data[0], axs[0,:], col, color='b', sample_size=None)

axs[0][0].set_yscale('log')
axs[1][0].set_yscale('log')
plt.legend(f+a, ['fpga', 'arista'], loc='upper right')

for exp, ax in zip(EXPERIMENTS, axs[0,:]):
    ax.set_title(exp)

axs[0][0].set_ylabel("\nLatency (s)")
axs[1][0].set_ylabel("1000 Samples\nLatency (s)")

plt.tight_layout()
fig, axs = plt.subplots(3, len(EXPERIMENTS), sharey=True, figsize=(7,5))
column = 'latency'
plot_group_stat(tofino_data[0], axs[0,:], column, start_time=10, stop_time=40)
plot_group_stat(fpga_data[0], axs[1,:], column, start_time=10, stop_time=40)
plot_group_stat(arista_data[0], axs[2,:], column, start_time=10, stop_time=40)
plt.yscale("log")
%matplotlib notebook
fig, axs = plt.subplots(3, len(EXPERIMENTS), squeeze=False, sharey=True, sharex=True, figsize=(7,5))

plot_group_stat(tofino_data[0], axs[0,:], start_time=5, stop_time=35)
axs[0][0].set_ylabel("Tofino (S)")

plot_group_stat(fpga_data[0], axs[1,:], start_time=5, stop_time=35)
axs[1][0].set_ylabel("FPGA (S)")

plot_group_stat(arista_data[0], axs[2,:], start_time=5, stop_time=35)
axs[2][0].set_ylabel("Arista (S)")


plt.suptitle("MCD Latency")
plt.xlim([0, 6])
axs[0][0].set_xticklabels([])
plt.tight_layout()
plt.subplots_adjust(top=.8)
plt.yscale('log')

%matplotlib notebook
fig, axs = plt.subplots(3, len(EXPERIMENTS), squeeze=False, sharey=True, sharex=True, figsize=(7,5))

plot_group_stat(tofino_data[1], axs[0,:], start_time=5, stop_time=35)
axs[0][0].set_ylabel("Tofino (%)")

plot_group_stat(fpga_data[1], axs[1,:], start_time=5, stop_time=35)
axs[1][0].set_ylabel("FPGA (%)")

plot_group_stat(arista_data[1], axs[2,:], start_time=5, stop_time=35)
axs[2][0].set_ylabel("Arista (%)")


plt.suptitle("MCD Latency")
plt.xlim([0, 6])
axs[0][0].set_xticklabels([])
plt.tight_layout()
plt.subplots_adjust(top=.8)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13735927269301426, 'fraction_numerical': 0.028451646332846577, 'mean_word_length': 3.555793991416309, 'pattern_counts': {'":': 0, '<': 50, '<?xml version=': 0, '>': 48, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '11', 'dir': 'jupyter-structured-clean-dedup', 'id': '128470', 'license': 'Apache-2.0', 'n_tokens_mistral': 7834, 'n_tokens_neox': 7421, 'n_words': 1655, 'path': 'testing/e2e/analysis/mcd_data_loading.ipynb', 'repo_name': 'CSEYJ/Flightplan', 'size': '21229'}