ZMQ Frontend + Caching Improvements

CMakeLists.txt

@@ -25,10 +25,18 @@ find_package(Folly REQUIRED)
 add_library(folly INTERFACE IMPORTED)
 set_property(TARGET folly PROPERTY
     INTERFACE_INCLUDE_DIRECTORIES ${FOLLY_INCLUDE_DIR})
-set(FOLLY_LINK_FLAGS "-lglog")
+set(FOLLY_LINK_FLAGS "-lglog -lcrypto -lssl")
 set_property(TARGET folly PROPERTY
     INTERFACE_LINK_LIBRARIES ${FOLLY_LINK_FLAGS} ${FOLLY_LIBRARY})
 
+# Include Wangle as an imported target
+find_package(Wangle REQUIRED)
+add_library(wangle INTERFACE IMPORTED)
+set_property(TARGET wangle PROPERTY
+        INTERFACE_INCLUDE_DIRECTORIES ${WANGLE_INCLUDE_DIR})
+set_property(TARGET wangle PROPERTY
+        INTERFACE_LINK_LIBRARIES ${WANGLE_LIBRARY})
+
 # set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -pedantic -Wextra -Werror -Wno-deprecated-declarations -Wno-unsupported-friend")
 set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -Wextra -Wno-deprecated-declarations -Wno-unused-parameter -Wno-sign-compare")
 if (CMAKE_CXX_COMPILER_ID MATCHES "Clang")

clipper_admin/clipper_manager.py

@@ -37,6 +37,7 @@
 
 DEFAULT_REDIS_IP = "redis"
 DEFAULT_REDIS_PORT = 6379
+DEFAULT_PREDICTION_CACHE_SIZE_BYTES = 33554432L
 CLIPPER_QUERY_PORT = 1337
 CLIPPER_MANAGEMENT_PORT = 1338
 CLIPPER_RPC_PORT = 7000
@@ -104,7 +105,9 @@ class Clipper:
     restart_containers : bool, optional
         If true, containers will restart on failure. If false, containers
         will not restart automatically.
-
+    cache_size : int, optional
+        The size of Clipper's prediction cache in bytes. Default cache size is
+        32 MiB
     """
 
     def __init__(self,
@@ -118,7 +121,8 @@ def __init__(self,
                  redis_port=DEFAULT_REDIS_PORT,
                  redis_persistence_path=None,
                  restart_containers=False,
-                 rpc_frontend=False):
+                 rpc_frontend=False,
+                 cache_size=DEFAULT_PREDICTION_CACHE_SIZE_BYTES):
         self.redis_ip = redis_ip
         self.redis_port = redis_port
         if rpc_frontend:
@@ -152,7 +156,8 @@ def __init__(self,
                 'query_frontend': {
                     'command': [
                         '--redis_ip=%s' % self.redis_ip,
-                        '--redis_port=%d' % self.redis_port
+                        '--redis_port=%d' % self.redis_port,
+                        '--prediction_cache_size=%d' % cache_size
                     ],
                     'depends_on': ['mgmt_frontend'],
                     'image':
@@ -1520,6 +1525,7 @@ def _publish_new_model(self, name, version, labels, input_type,
         headers = {'Content-type': 'application/json'}
         r = requests.post(url, headers=headers, data=req_json)
         if r.status_code == requests.codes.ok:
+            print(r.text)
             return True
         else:
             print("Error publishing model: %s" % r.text)

cmake/FindFolly.cmake

@@ -15,8 +15,8 @@ CMAKE_MINIMUM_REQUIRED(VERSION 2.8.7 FATAL_ERROR)
 
 INCLUDE(FindPackageHandleStandardArgs)
 
-FIND_LIBRARY(FOLLY_LIBRARY folly)
-FIND_PATH(FOLLY_INCLUDE_DIR "folly/String.h")
+FIND_LIBRARY(FOLLY_LIBRARY folly PATHS $ENV{FOLLY_ROOT}/lib $ENV{PATH})
+FIND_PATH(FOLLY_INCLUDE_DIR "folly/String.h" HINTS $ENV{FOLLY_ROOT}/include)
 
 SET(FOLLY_LIBRARIES ${FOLLY_LIBRARY})
 

cmake/FindWangle.cmake

@@ -0,0 +1,35 @@
+#
+# - Try to find Facebook wangle library
+# This will define
+# WANGLE_FOUND
+# WANGLE_INCLUDE_DIR
+# WANGLE_LIBRARIES
+#
+
+find_package(Folly REQUIRED)
+
+find_path(
+    WANGLE_INCLUDE_DIR
+    NAMES "wangle/channel/Pipeline.h"
+    HINTS
+        "/usr/local/facebook/include"
+)
+
+find_library(
+    WANGLE_LIBRARY
+    NAMES wangle
+    HINTS
+        "/usr/local/facebook/lib"
+)
+
+set(WANGLE_LIBRARIES ${WANGLE_LIBRARY} ${FOLLY_LIBRARIES})
+
+include(FindPackageHandleStandardArgs)
+find_package_handle_standard_args(
+    WANGLE DEFAULT_MSG WANGLE_INCLUDE_DIR WANGLE_LIBRARIES)
+
+mark_as_advanced(WANGLE_INCLUDE_DIR WANGLE_LIBRARIES WANGLE_FOUND)
+
+if(WANGLE_FOUND AND NOT WANGLE_FIND_QUIETLY)
+    message(STATUS "WANGLE: ${WANGLE_INCLUDE_DIR}")
+endif()

grpc_client_py/grpc_benchmark.py

@@ -0,0 +1,64 @@
+from __future__ import print_function
+
+import sys
+import grpc
+import logging
+import numpy as np
+import time
+
+import clipper_frontend_pb2
+import clipper_frontend_pb2_grpc
+
+from datetime import datetime
+from multiprocessing import Process
+
+DATA_TYPE_BYTES = 0
+DATA_TYPE_INTS = 1
+DATA_TYPE_FLOATS = 2
+DATA_TYPE_DOUBLES = 3
+DATA_TYPE_STRINGS = 4
+
+CIFAR_SIZE_DOUBLES = 384
+
+input_type = "doubles"
+app_name = "app1"
+model_name = "m1"
+
+def run(proc_num):
+	channel = grpc.insecure_channel('localhost:1337')
+	stub = clipper_frontend_pb2_grpc.PredictStub(channel)
+	i = 0
+	latency = 0
+	file_name = "/tmp/bench_{}".format(proc_num)
+	while True:
+		begin = datetime.now()
+		x = clipper_frontend_pb2.DoubleData(data=list(np.random.random(CIFAR_SIZE_DOUBLES)))
+		req = clipper_frontend_pb2.PredictRequest(application=app_name, data_type=DATA_TYPE_DOUBLES, double_data=x)
+		response = stub.Predict(req)
+		print("Received response!")
+		end = datetime.now()
+
+		latency += (end - begin).total_seconds()
+
+		if i > 0 and i % 100 == 0:
+			print("Throughput: {} qps\n".format(float(latency) / i))
+			i = 0
+			latency = 0
+
+		i += 1
+
+if __name__ == "__main__":
+	if len(sys.argv) < 2:
+		raise
+
+	num_procs = int(sys.argv[1])
+
+	processes = []
+
+	for i in range(num_procs):
+		p = Process(target=run, args=('%d'.format(i),))
+		p.start()
+		processes.append(p)
+
+	for p in processes:
+		p.join()

src/CMakeLists.txt

@@ -6,6 +6,7 @@ add_subdirectory(libs)
 add_subdirectory(libclipper)
 add_subdirectory(rest_frontend)
 add_subdirectory(rpc_frontend)
+add_subdirectory(zmq_frontend)
 add_subdirectory(management)
 add_subdirectory(benchmarks)
 add_subdirectory(container)

src/benchmarks/src/end_to_end_bench.cpp

@@ -188,8 +188,9 @@ void run_benchmark(std::unordered_map<std::string, std::string> &config) {
       static_cast<char *>(malloc(DEFAULT_OUTPUT.size())), free);
   memcpy(default_output_content.get(), DEFAULT_OUTPUT.data(),
          DEFAULT_OUTPUT.size());
+
   clipper::Output parsed_default_output(
-      std::make_tuple(default_output_content, 0, DEFAULT_OUTPUT.size()), {});
+      std::make_shared<StringOutput>(default_output_content, 0, DEFAULT_OUTPUT.size()), {});
   auto init_state = p.init_state(parsed_default_output);
   clipper::StateKey state_key{TEST_APPLICATION_LABEL, clipper::DEFAULT_USER_ID,
                               0};