Commit c8834a73 authored by Zhihao Bai's avatar Zhihao Bai

NetCache

parent 735f0f1f
================================================= 0 Introduction ==================================================
In this repository, I implemented a simple NetCache with standard P4 language and designed an experiemnt with behavior model to show the efficiency of NetCache.
I create a network with mininet, containing 1 switch and 3 hosts. One host is the server, which handles READ queries. One host is the client, which sends READ queries. The last host is to simulate the controller of the programmable switch, because the behavior model does not provide such an interface.
The experiemnts runs in the following steps. First, the switch starts. Some table entries are added. Second, the server starts. The server loads pre-generated key-value items. Third, the controller starts. The controller sends some UPDATE queries to the server to get values of some pre-determined hot items, and insert them to the switch. Finally, the client starts. The client sends READ queries to the server and receives replies. If a query hits the cache in the switch, the switch would directly reply this query without routing it to the server.
If some uncached items are detected to be hot by the heavy-hitter, the switch would send a HOT_READ to the controller.
================================================= 1 Obtaining required software ==================================================
It is recommended to do the following things in the directory "NetCache/".
Firstly, you need to get the p4 compiler from Github, and install required dependencies.
git clone https://github.com/p4lang/p4c-bm.git p4c-bmv2
cd p4c-bmv2
sudo pip install -r requirements.txt
Secondly, you need to get the behavior model of p4 from github, install dependencies and compile the behavior model.
git clone https://github.com/p4lang/behavioral-model.git bmv2
cd bmv2
install_deps.sh
./autogen.sh
./configure
make
Finally, you need to install some other tools which are used in this simulation.
sudo apt-get install mininet python-ipaddr
sudo pip install scapy thrift networkx
If you do not do the above things in "NetCache/", you need to modify the path to p4c-bmv2 and bmv2 in NetCache/mininet/run_demo.sh.
================================================= 2 Content ==================================================
NetCache/generator: Python programs, which generates key-value items and queries.
NetCache/client: Two Python programs for the client. "send.py" can read queries from the "query.txt" file and send queries to the server. "receive.py" can receive replies from the server and the switch. In addition, both of these programs can print current READ throughput to the screen.
NetCache/server: One Python program for the server. "server.py" can read key-value items from the "kv.txt" file, and reply UPDATE queries from the controller and READ queries from the client. In addition, this program can print current READ throughput to the screen.
NetCache/p4src: Codes for the NetCache in standard P4 language.
NetCache/controller: One Python program for the controller. "controller.py" can read hot keys from the "hot.txt" file, and send UPDATE requests to the server. Then the switch would insert values to the cache when detect UDPATE replies from the server. After updating the cache of the switch, the controller would wait for "HOT_READ" packets, which shows that a key is detected as hot. In addition, this program can print HOT_READ reports with heavy hitter information to the screen.
NetCache/mininet: Scripts to run the experiments.
================================================= 3 Run Simulation ==================================================
Experiment configuration: IP address "10.0.0.1" is for the client. IP address "10.0.0.2" is for the server. IP address "10.0.0.3" is for the controller. There are 1000 key-value items in total, following zipf-0.90 distribution. Items whose keys are 1, 3, 5, ..., 99 will be inserted to the cache of the switch, and items whose keys are 2, 4, ..., 100 will be detected as hot items and reported to the controller after running for several seconds. If an uncached item is accessed for 128 times, it would be reported to the controller, and this parameter can be changed in "NetCache/p4src/heavy_hitter.p4".
Before the experiment starts, you need to generate some files. Run "python gen_kv.py" in "NetCache/generator", and you will get "kv.txt" and "hot.txt". Copy "kv.txt" to "NetCache/server" and copy "hot.txt" to "NetCache/controller". Then run "python gen_query_zipf.py" in "NetCache/generator", and you will get "query.txt". It takes several minutes. Then copy "query.txt" to "NetCache/client".
To initialize the environment, open a terminal in "NetCache/mininet", and run "./run_demo.sh". After you can see "Ready! Starting CLI: mininet>", you can begin to run the experiment. In the following description, I will call this terminal "mininet terminal".
Firstly, in the mininet terminal, run "xterm h2" to open a terminal for the server. In the new terminal, enter "NetCache/server" by running "cd ../server" and run "python server.py". Then the server starts. You can see two numbers, which are the number of READ queries in one second and the total number of READ queries in the past time.
Secondly, in the mininet terminal, run "xterm h3" to open a terminal for the controller. In the new terminal, enter "NetCache/controller" by running "cd ../controller" and run "python controller.py". Then the controller starts. When the controller receives a HOT_READ report, the detected key and values of the heavy hitter will be displayed.
Thidly, in the mininet terminal, run "xterm h1" to open a terminal for "receive.py" of the client. In the new terminal, enter "NetCache/client" by running "cd ../client" and run "python receive.py". Then you can see the number of READ replies received per second and the total number of READ replies in the past time.
Finally, in the mininet terminal, run "xterm h1" to open a terminal for "send.py" of the client. In the new terminal, enter "NetCache/client" by running "cd ../client" and run "python send.py". THen you can the number of READ replies received per second and the total number of READ replies in the past time. At the same time, the displayed numbers of the server and the "receive.py" will change with the "send.py", and after several seconds the controller will show the detected hot keys.
In addition, READ replies received by the "receive.py" is more than READ requests handled by the server. This is because some queries are handled by the switch.
NC_READ_REQUEST = 0
NC_READ_REPLY = 1
NC_HOT_READ_REQUEST = 2
NC_WRITE_REQUEST = 4
NC_WRITE_REPLY = 5
NC_UPDATE_REQUEST = 8
NC_UPDATE_REPLY = 9
import socket
import struct
import time
import thread
from nc_config import *
NC_PORT = 8888
CLIENT_IP = "10.0.0.1"
SERVER_IP = "10.0.0.2"
CONTROLLER_IP = "10.0.0.3"
path_reply = "reply.txt"
len_key = 16
counter = 0
def counting():
last_counter = 0
while True:
print (counter - last_counter), counter
last_counter = counter
time.sleep(1)
thread.start_new_thread(counting, ())
#f = open(path_reply, "w")
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.bind((CLIENT_IP, NC_PORT))
while True:
packet, addr = s.recvfrom(1024)
counter = counter + 1
#op = struct.unpack("B", packet[0])
#key_header = struct.unpack(">I", packet[1:5])[0]
#f.write(str(op) + ' ')
#f.write(str(key_header) + '\n')
#f.flush()
#print counter
#f.close()
import socket
import struct
import time
import thread
from nc_config import *
NC_PORT = 8888
CLIENT_IP = "10.0.0.1"
SERVER_IP = "10.0.0.2"
CONTROLLER_IP = "10.0.0.3"
path_query = "query.txt"
query_rate = 1000
len_key = 16
counter = 0
def counting():
last_counter = 0
while True:
print (counter - last_counter), counter
last_counter = counter
time.sleep(1)
thread.start_new_thread(counting, ())
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
f = open(path_query, "r")
interval = 1.0 / (query_rate + 1)
for line in f.readlines():
line = line.split()
op = line[0]
key_header = int(line[1])
key_body = line[2:]
op_field = struct.pack("B", NC_READ_REQUEST)
key_field = struct.pack(">I", key_header)
for i in range(len(key_body)):
key_field += struct.pack("B", int(key_body[i], 16))
packet = op_field + key_field
s.sendto(packet, (SERVER_IP, NC_PORT))
counter = counter + 1
time.sleep(interval)
f.close()
import socket
import struct
import time
import thread
from nc_config import *
NC_PORT = 8888
CLIENT_IP = "10.0.0.1"
SERVER_IP = "10.0.0.2"
CONTROLLER_IP = "10.0.0.3"
path_hot = "hot.txt"
path_log = "controller_log.txt"
len_key = 16
len_val = 128
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.bind((CONTROLLER_IP, NC_PORT))
## Initiate the switch
op = NC_UPDATE_REQUEST
op_field = struct.pack("B", op)
f = open(path_hot, "r")
for line in f.readlines():
line = line.split()
key_header = line[0]
key_body = line[1:]
key_header = int(key_header)
for i in range(len(key_body)):
key_body[i] = int(key_body[i], 16)
key_field = ""
key_field += struct.pack(">I", key_header)
for i in range(len(key_body)):
key_field += struct.pack("B", key_body[i])
packet = op_field + key_field
s.sendto(packet, (SERVER_IP, NC_PORT))
time.sleep(0.001)
f.close()
## Listen hot report
#f = open(path_log, "w")
while True:
packet, addr = s.recvfrom(2048)
op_field = packet[0]
key_field = packet[1:len_key + 1]
load_field = packet[len_key + 1:]
op = struct.unpack("B", op_field)[0]
if (op != NC_HOT_READ_REQUEST):
continue
key_header = struct.unpack(">I", key_field[:4])[0]
load = struct.unpack(">IIII", load_field)
counter = counter + 1
print "\tHot Item:", key_header, load
#f.write(str(key_header) + ' ')
#f.write(str(load) + ' ')
#f.write("\n")
#f.flush()
#f.close()
NC_READ_REQUEST = 0
NC_READ_REPLY = 1
NC_HOT_READ_REQUEST = 2
NC_WRITE_REQUEST = 4
NC_WRITE_REPLY = 5
NC_UPDATE_REQUEST = 8
NC_UPDATE_REPLY = 9
import random
path_kv = "kv.txt" #The path to save generated keys and values
path_hot = "hot.txt" #The path to save the hot keys
len_key = 16 #The number of bytes in the key
len_val = 128 #The number of bytes in the value
max_key = 1000 #The number of keys
max_hot = 100 #The number of hot keys
f = open(path_kv, "w")
f_hot = open(path_hot, "w")
f.write(str(max_key) + "\n\n")
for i in range(1, max_key + 1):
## Generate a key-value item
#Select a key
key_header = i
key_body = [0] * (len_key - 4)
#Select a value
val = [1] * len_val #The value
###################################################################################################
## Output the key and the value to the file
f.write(str(key_header) + " ")
for i in range(len(key_body)):
f.write(hex(key_body[i]) + " ")
f.write("\n")
for i in range(len(val)):
f.write(hex(val[i]) + " ")
f.write("\n\n")
###################################################################################################
##Output the hot key to the file
if (key_header <= max_hot and key_header % 2 == 1):
f_hot.write(str(key_header) + " ")
for i in range(len(key_body)):
f_hot.write(hex(key_body[i]) + " ")
f_hot.write("\n")
###################################################################################################
f.flush()
f.close()
f_hot.flush()
f_hot.close()
import random
path_query = "query.txt"
num_query = 1000000
len_key = 16
len_val = 128
max_key = 1000
f = open(path_query, "w")
for i in range(num_query):
#Randomly select a key
key_header = random.randint(1, max_key)
key_body = [0] * (len_key - 4)
#Save the generated query to the file
f.write("get ")
f.write(str(key_header) + ' ')
for i in range(len(key_body)):
f.write(hex(key_body[i]) + ' ')
f.write('\n')
f.flush()
f.close()
import random
import math
path_query = "query.txt"
num_query = 1000000
zipf = 0.99
len_key = 16
len_val = 128
max_key = 1000
#Zipf
zeta = [0.0]
for i in range(1, max_key + 1):
zeta.append(zeta[i - 1] + 1 / pow(i, zipf))
field = [0] * (num_query + 1)
k = 1
for i in range(1, num_query + 1):
if (i > num_query * zeta[k] / zeta[max_key]):
k = k + 1
field[i] = k
#Generate queries
f = open(path_query, "w")
for i in range(num_query):
#Randomly select a key in zipf distribution
r = random.randint(1, num_query)
key_header = field[r]
key_body = [0] * (len_key - 4)
#Save the generated query to the file
f.write("get ")
f.write(str(key_header) + ' ')
for i in range(len_key - 4):
f.write(hex(key_body[i]) + ' ')
f.write('\n')
f.flush()
f.close()
path_to_cmd = "commands_cache.txt"
max_hot = 100
len_key = 16
f = open(path_to_cmd, "w")
for i in range(1, max_hot + 1, 2):
x = i << ((len_key - 4) * 8)
f.write("table_add check_cache_exist check_cache_exist_act %d => %d\n" % (x, i))
f.flush()
f.close()
path_to_cmd = "commands_value.txt"
f = open(path_to_cmd, "w")
for i in range(1, 9):
for j in range(1, 5):
f.write("table_set_default read_value_%d_%d read_value_%d_%d_act\n" % (i, j, i, j))
f.write("table_set_default add_value_header_%d add_value_header_%d_act\n" % (i, i))
f.write("table_set_default write_value_%d_%d write_value_%d_%d_act\n" % (i, j, i, j))
f.write("table_set_default remove_value_header_%d remove_value_header_%d_act\n" % (i, i))
f.flush()
f.close()
This diff is collapsed.
This diff is collapsed.
# Copyright 2013-present Barefoot Networks, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from mininet.net import Mininet
from mininet.node import Switch, Host
from mininet.log import setLogLevel, info, error, debug
from mininet.moduledeps import pathCheck
from sys import exit
import os
import tempfile
import socket
class P4Host(Host):
def config(self, **params):
r = super(Host, self).config(**params)
self.defaultIntf().rename("eth0")
for off in ["rx", "tx", "sg"]:
cmd = "/sbin/ethtool --offload eth0 %s off" % off
self.cmd(cmd)
# disable IPv6
self.cmd("sysctl -w net.ipv6.conf.all.disable_ipv6=1")
self.cmd("sysctl -w net.ipv6.conf.default.disable_ipv6=1")
self.cmd("sysctl -w net.ipv6.conf.lo.disable_ipv6=1")
return r
def describe(self):
print "**********"
print self.name
print "default interface: %s\t%s\t%s" %(
self.defaultIntf().name,
self.defaultIntf().IP(),
self.defaultIntf().MAC()
)
print "**********"
class P4Switch(Switch):
"""P4 virtual switch"""
device_id = 0
def __init__(self, name, sw_path = None, json_path = None,
thrift_port = None,
pcap_dump = False,
log_console = False,
verbose = False,
device_id = None,
enable_debugger = False,
**kwargs):
Switch.__init__(self, name, **kwargs)
assert(sw_path)
assert(json_path)
# make sure that the provided sw_path is valid
pathCheck(sw_path)
# make sure that the provided JSON file exists
if not os.path.isfile(json_path):
error("Invalid JSON file.\n")
exit(1)
self.sw_path = sw_path
self.json_path = json_path
self.verbose = verbose
logfile = "/tmp/p4s.{}.log".format(self.name)
self.output = open(logfile, 'w')
self.thrift_port = thrift_port
self.pcap_dump = pcap_dump
self.enable_debugger = enable_debugger
self.log_console = log_console
if device_id is not None:
self.device_id = device_id
P4Switch.device_id = max(P4Switch.device_id, device_id)
else:
self.device_id = P4Switch.device_id
P4Switch.device_id += 1
self.nanomsg = "ipc:///tmp/bm-{}-log.ipc".format(self.device_id)
@classmethod
def setup(cls):
pass
def check_switch_started(self, pid):
"""While the process is running (pid exists), we check if the Thrift
server has been started. If the Thrift server is ready, we assume that
the switch was started successfully. This is only reliable if the Thrift
server is started at the end of the init process"""
while True:
if not os.path.exists(os.path.join("/proc", str(pid))):
return False
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(0.5)
result = sock.connect_ex(("localhost", self.thrift_port))
if result == 0:
return True
def start(self, controllers):
"Start up a new P4 switch"
info("Starting P4 switch {}.\n".format(self.name))
args = [self.sw_path]
for port, intf in self.intfs.items():
if not intf.IP():
args.extend(['-i', str(port) + "@" + intf.name])
if self.pcap_dump:
args.append("--pcap")
# args.append("--useFiles")
if self.thrift_port:
args.extend(['--thrift-port', str(self.thrift_port)])
if self.nanomsg:
args.extend(['--nanolog', self.nanomsg])
args.extend(['--device-id', str(self.device_id)])
P4Switch.device_id += 1
args.append(self.json_path)
if self.enable_debugger:
args.append("--debugger")
if self.log_console:
args.append("--log-console")
logfile = "/tmp/p4s.{}.log".format(self.name)
info(' '.join(args) + "\n")
pid = None
with tempfile.NamedTemporaryFile() as f:
# self.cmd(' '.join(args) + ' > /dev/null 2>&1 &')
self.cmd(' '.join(args) + ' >' + logfile + ' 2>&1 & echo $! >> ' + f.name)
pid = int(f.read())
debug("P4 switch {} PID is {}.\n".format(self.name, pid))
if not self.check_switch_started(pid):
error("P4 switch {} did not start correctly.\n".format(self.name))
exit(1)
info("P4 switch {} has been started.\n".format(self.name))
def stop(self):
"Terminate P4 switch."
self.output.flush()
self.cmd('kill %' + self.sw_path)
self.cmd('wait')
self.deleteIntfs()
def attach(self, intf):
"Connect a data port"
assert(0)
def detach(self, intf):
"Disconnect a data port"
assert(0)
#!/bin/bash
# Copyright 2013-present Barefoot Networks, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
BMV2_PATH=../bmv2
P4C_BM_PATH=../p4c-bmv2
P4C_BM_SCRIPT=$P4C_BM_PATH/p4c_bm/__main__.py
SWITCH_PATH=$BMV2_PATH/targets/simple_switch/simple_switch
#CLI_PATH=$BMV2_PATH/tools/runtime_CLI.py
CLI_PATH=$BMV2_PATH/targets/simple_switch/sswitch_CLI
$P4C_BM_SCRIPT ../p4src/netcache.p4 --json netcache.json
# This gives libtool the opportunity to "warm-up"
sudo $SWITCH_PATH >/dev/null 2>&1
sudo PYTHONPATH=$PYTHONPATH:$BMV2_PATH/mininet/ python topo.py \
--behavioral-exe $SWITCH_PATH \
--json netcache.json \
--cli $CLI_PATH
#!/usr/bin/python
# Copyright 2013-present Barefoot Networks, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from mininet.net import Mininet
from mininet.topo import Topo
from mininet.log import setLogLevel, info
from mininet.cli import CLI
from mininet.link import TCLink
from p4_mininet import P4Switch, P4Host
import argparse
from time import sleep
import os
import subprocess
_THIS_DIR = os.path.dirname(os.path.realpath(__file__))
_THRIFT_BASE_PORT = 22222
parser = argparse.ArgumentParser(description='Mininet demo')
parser.add_argument('--behavioral-exe', help='Path to behavioral executable',
type=str, action="store", required=True)
parser.add_argument('--json', help='Path to JSON config file',
type=str, action="store", required=True)
parser.add_argument('--cli', help='Path to BM CLI',
type=str, action="store", required=True)
args = parser.parse_args()
class MyTopo(Topo):
def __init__(self, sw_path, json_path, nb_hosts, nb_switches, links, **opts):
# Initialize topology and default options
Topo.__init__(self, **opts)
for i in xrange(nb_switches):
switch = self.addSwitch('s%d' % (i + 1),
sw_path = sw_path,
json_path = json_path,
thrift_port = _THRIFT_BASE_PORT + i,
pcap_dump = True,
device_id = i)
for h in xrange(nb_hosts):
host = self.addHost('h%d' % (h + 1))
for a, b in links:
self.addLink(a, b)
def read_topo():
nb_hosts = 0
nb_switches = 0
links = []
with open("topo.txt", "r") as f:
line = f.readline()[:-1]
w, nb_switches = line.split()
assert(w == "switches")
line = f.readline()[:-1]
w, nb_hosts = line.split()
assert(w == "hosts")
for line in f:
if not f: break
a, b = line.split()
links.append( (a, b) )
return int(nb_hosts), int(nb_switches), links
def main():
nb_hosts, nb_switches, links = read_topo()
topo = MyTopo(args.behavioral_exe,
args.json,
nb_hosts, nb_switches, links)
net = Mininet(topo = topo,
host = P4Host,
switch = P4Switch,
controller = None,
autoStaticArp=True )
net.start()
for n in range(nb_hosts):
h = net.get('h%d' % (n + 1))
for off in ["rx", "tx", "sg"]:
cmd = "/sbin/ethtool --offload eth0 %s off" % off
print cmd
h.cmd(cmd)
print "disable ipv6"
h.cmd("sysctl -w net.ipv6.conf.all.disable_ipv6=1")
h.cmd("sysctl -w net.ipv6.conf.default.disable_ipv6=1")
h.cmd("sysctl -w net.ipv6.conf.lo.disable_ipv6=1")
h.cmd("sysctl -w net.ipv4.tcp_congestion_control=reno")
h.cmd("iptables -I OUTPUT -p icmp --icmp-type destination-unreachable -j DROP")
h.setIP("10.0.0.%d" % (n + 1))
h.setMAC("aa:bb:cc:dd:ee:0%d" % (n + 1))
for i in range(nb_hosts):
if (i != n):
h.setARP("10.0.0.%d" % (i + 1), "aa:bb:cc:dd:ee:0%d" % (i + 1))
net.get('s1').setMAC("aa:bb:cc:dd:ee:1%d" % (n + 1), "s1-eth%d" % (n + 1))
sleep(1)
for i in range(nb_switches):
#cmd = [args.cli, "--json", args.json, "--thrift-port", str(_THRIFT_BASE_PORT + i)]
cmd = [args.cli, args.json, str(_THRIFT_BASE_PORT + i)]
with open("commands.txt", "r") as f:
print " ".join(cmd)
try:
output = subprocess.check_output(cmd, stdin = f)
print output
except subprocess.CalledProcessError as e:
print e
print e.output
sleep(1)
print "Ready !"
CLI( net )
net.stop()
if __name__ == '__main__':
setLogLevel( 'info' )
main()
switches 1
hosts 3
h1 s1
h2 s1
h3 s1
NC_READ_REQUEST = 0
NC_READ_REPLY = 1
NC_HOT_READ_REQUEST = 2
NC_WRITE_REQUEST = 4
NC_WRITE_REPLY = 5
NC_UPDATE_REQUEST = 8
NC_UPDATE_REPLY = 9
header_type nc_cache_md_t {
fields {
cache_exist: 1;
cache_index: 14;
cache_valid: 1;
}
}
metadata nc_cache_md_t nc_cache_md;
action check_cache_exist_act(index) {
modify_field (nc_cache_md.cache_exist, 1);
modify_field (nc_cache_md.cache_index, index);
}
table check_cache_exist {
reads {
nc_hdr.key: exact;