gRPC and Protobuf support for remote user-defined operations

.github/scripts/Dockerfile.checkin

@@ -21,7 +21,7 @@ ENV DEBIAN_FRONTEND=noninteractive
 ENV DEBCONF_NOWARNINGS="yes"
 ENV PYTHON_BASE="3.12"
 ENV PYTHON_VERSION="${PYTHON_BASE}.3"
-ENV PROTOBUF_VERSION="24.2"
+ENV PROTOBUF_VERSION="31.0"
 ENV NUMPY_MIN_VERSION="1.26.0"
 ENV VIRTUAL_ENV=/opt/venv
 
@@ -81,7 +81,8 @@ ENV AUTOCONF_VERSION="2.71" \
     OPENCV_VERSION="4.9.0" \
     PEG_VERSION="0.1.19" \
     TILEDB_VERSION="2.14.1" \
-    VALIJSON_VERSION="v0.6"
+    VALIJSON_VERSION="v0.6" \
+    GRPC_VERSION="v1.73.0"
 
 # CMAKE
 # hadolint ignore=DL3003,SC2086
@@ -91,18 +92,26 @@ RUN git clone --branch ${CMAKE_VERSION} https://github.com/Kitware/CMake.git /de
 
 # PROTOBUF & ITS DEPENDENCIES
 # hadolint ignore=DL3003,SC2086
-RUN git clone -b "v${PROTOBUF_VERSION}" --recurse-submodules https://github.com/protocolbuffers/protobuf.git /dependencies/protobuf && \
-        cd /dependencies/protobuf/third_party/googletest && mkdir build && cd build/ && \
+
+# googletest
+RUN git clone https://github.com/google/googletest.git -b v1.12.0 /dependencies/googletest && \
+            cd /dependencies/googletest && mkdir build && cd build && \
             cmake -DCMAKE_POSITION_INDEPENDENT_CODE=ON -DCMAKE_BUILD_TYPE=Release \
                 -DBUILD_SHARED_LIBS=ON -DCMAKE_INSTALL_PREFIX=/opt/dist/usr/local \
                 -DBUILD_GMOCK=ON -DCMAKE_CXX_STANDARD=17 .. && \
-            make ${BUILD_THREADS} && make install && \
-        cd /dependencies/protobuf/third_party/abseil-cpp && mkdir build && cd build && \
+            make ${BUILD_THREADS} && make install
+
+# Abseil-CPP
+RUN git clone https://github.com/abseil/abseil-cpp.git /dependencies/abseil && \
+            cd /dependencies/abseil && mkdir build && cd build && \
             cmake -DCMAKE_POSITION_INDEPENDENT_CODE=ON -DBUILD_SHARED_LIBS=ON \
                 -DCMAKE_INSTALL_PREFIX=/opt/dist/usr/local -DABSL_BUILD_TESTING=ON \
                 -DABSL_USE_EXTERNAL_GOOGLETEST=ON \
                 -DABSL_FIND_GOOGLETEST=ON -DCMAKE_CXX_STANDARD=17 .. && \
-            make ${BUILD_THREADS} && make install && ldconfig /opt/dist/usr/local/lib && \
+            make ${BUILD_THREADS} && make install && ldconfig /opt/dist/usr/local/lib
+
+# Protobuf
+RUN git clone -b "v${PROTOBUF_VERSION}" --recurse-submodules https://github.com/protocolbuffers/protobuf.git /dependencies/protobuf && \
         cd /dependencies/protobuf && \
             cmake -DCMAKE_POSITION_INDEPENDENT_CODE=ON -DCMAKE_INSTALL_PREFIX=/opt/dist/usr/local \
                 -DCMAKE_CXX_STANDARD=17 -Dprotobuf_BUILD_SHARED_LIBS=ON \
@@ -111,6 +120,33 @@ RUN git clone -b "v${PROTOBUF_VERSION}" --recurse-submodules https://github.com/
                 -Dabsl_DIR=/opt/dist/usr/local/lib/cmake/absl . && \
             make ${BUILD_THREADS} && make install
 
+# RUN git clone -b "v${PROTOBUF_VERSION}" --recurse-submodules https://github.com/protocolbuffers/protobuf.git /dependencies/protobuf && \
+#         cd /dependencies/protobuf/third_party/googletest && mkdir build && cd build/ && \
+#             cmake -DCMAKE_POSITION_INDEPENDENT_CODE=ON -DCMAKE_BUILD_TYPE=Release \
+#                 -DBUILD_SHARED_LIBS=ON -DCMAKE_INSTALL_PREFIX=/opt/dist/usr/local \
+#                 -DBUILD_GMOCK=ON -DCMAKE_CXX_STANDARD=17 .. && \
+#             make ${BUILD_THREADS} && make install && \
+#         cd /dependencies/protobuf/third_party/abseil-cpp && mkdir build && cd build && \
+#             cmake -DCMAKE_POSITION_INDEPENDENT_CODE=ON -DBUILD_SHARED_LIBS=ON \
+#                 -DCMAKE_INSTALL_PREFIX=/opt/dist/usr/local -DABSL_BUILD_TESTING=ON \
+#                 -DABSL_USE_EXTERNAL_GOOGLETEST=ON \
+#                 -DABSL_FIND_GOOGLETEST=ON -DCMAKE_CXX_STANDARD=17 .. && \
+#             make ${BUILD_THREADS} && make install && ldconfig /opt/dist/usr/local/lib && \
+#         cd /dependencies/protobuf && \
+#             cmake -DCMAKE_POSITION_INDEPENDENT_CODE=ON -DCMAKE_INSTALL_PREFIX=/opt/dist/usr/local \
+#                 -DCMAKE_CXX_STANDARD=17 -Dprotobuf_BUILD_SHARED_LIBS=ON \
+#                 -Dprotobuf_ABSL_PROVIDER=package \
+#                 -Dprotobuf_BUILD_TESTS=ON \
+#                 -Dabsl_DIR=/opt/dist/usr/local/lib/cmake/absl . && \
+#             make ${BUILD_THREADS} && make install
+
+# gRPC
+RUN git clone -b ${GRPC_VERSION} https://github.com/grpc/grpc && \
+        cd grpc && \
+        git submodule update --init && mkdir -p cmake/build && cd cmake/build && \
+        cmake -DCMAKE_CXX_STANDARD=17 -DgRPC_INSTALL=ON ../.. && \
+        make ${BUILD_THREADS} && make install
+
 # OPENCV
 # hadolint ignore=DL3003,SC2086
 RUN git clone https://github.com/opencv/opencv.git /dependencies/opencv && \
@@ -239,6 +275,9 @@ COPY --from=build /usr/include/libwebsocket[s] /usr/include/libwebsockets
 COPY --from=build /usr/local/lib/libkubernetes.s[o] /usr/local/lib/libkubernetes.so
 COPY --from=build /usr/local/lib/libyaml.s[o] /usr/local/lib/libyaml.so
 COPY --from=build /usr/lib/x86_64-linux-gnu/libwebsockets.s[o] /usr/lib/x86_64-linux-gnu/libwebsockets.so
+COPY --from=build /usr/local/include/ /usr/local/include/
+COPY --from=build /usr/local/bin /usr/local/bin
+COPY --from=build /usr/local/lib/ /usr/local/lib/
 ENV PATH="$VIRTUAL_ENV/bin:$PATH"
 
 # hadolint ignore=DL3008,SC2086
@@ -253,7 +292,7 @@ RUN apt-get update -y && apt-get upgrade -y && \
     apt-get --purge remove -y python3.11 && apt-get autoremove -y && \
     apt-get clean && rm -rf /var/lib/apt/lists/* && \
     echo "/usr/local/lib" >> /etc/ld.so.conf.d/all-libs.conf && ldconfig && \
-    python3 -m pip install --no-cache-dir "numpy>=${NUMPY_MIN_VERSION},<2.0.0" "protobuf==4.${PROTOBUF_VERSION}" \
+    python3 -m pip install --no-cache-dir "numpy>=${NUMPY_MIN_VERSION},<2.0.0" "protobuf==6.${PROTOBUF_VERSION}" \
         "coverage>=7.3.1" "cryptography>=44.0.1"
 
 # COVERAGE TESTING

.gitignore

@@ -13,3 +13,6 @@ tmp
 
 # Jetbrains Specific
 .idea
+
+# Remote UDF Temps
+remote_function/entity_pb2*

CMakeLists.txt

@@ -22,6 +22,13 @@ find_package( OpenCV REQUIRED )
 find_package(Protobuf CONFIG REQUIRED)
 find_package( CURL REQUIRED )
 find_package(AWSSDK REQUIRED COMPONENTS core s3)
+find_package(gRPC CONFIG REQUIRED)
+# find_package(absl CONFIG REQUIRED)
+find_package(GTest REQUIRED)
+
+# set(_PROTOBUF_PROTOC $<TARGET_FILE:protobuf::protoc>)
+# set(_GRPC_GRPCPP gRPC::grpc++)
+# set(_GRPC_CPP_PLUGIN_EXECUTABLE $<TARGET_FILE:gRPC::grpc_cpp_plugin>)
 
 include_directories(${Protobuf_INCLUDE_DIRS})
 include_directories(${CMAKE_CURRENT_BINARY_DIR})
@@ -35,8 +42,9 @@ add_library(vdms_protobuf OBJECT
   ${CMAKE_CURRENT_SOURCE_DIR}/utils/src/protobuf/partitionerMessages.proto
   ${CMAKE_CURRENT_SOURCE_DIR}/utils/src/protobuf/pmgdMessages.proto
   ${CMAKE_CURRENT_SOURCE_DIR}/utils/src/protobuf/queryMessage.proto
+  ${CMAKE_CURRENT_SOURCE_DIR}/utils/src/protobuf/entity.proto
 )
-target_link_libraries(vdms_protobuf PUBLIC protobuf::libprotobuf)
+target_link_libraries(vdms_protobuf PUBLIC protobuf::libprotobuf gRPC::grpc++)
 set(PROTO_BINARY_DIR ${CMAKE_CURRENT_BINARY_DIR})
 target_include_directories(vdms_protobuf PUBLIC "$<BUILD_INTERFACE:${PROTO_BINARY_DIR}>")
 protobuf_generate(
@@ -46,6 +54,14 @@ protobuf_generate(
   PROTOC_OUT_DIR "${PROTO_BINARY_DIR}"
 )
 
+protobuf_generate(
+    TARGET vdms_protobuf
+    LANGUAGE grpc
+    GENERATE_EXTENSIONS .grpc.pb.h .grpc.pb.cc
+    PLUGIN "protoc-gen-grpc=\$<TARGET_FILE:gRPC::grpc_cpp_plugin>"
+    IMPORT_DIRS "${CMAKE_CURRENT_SOURCE_DIR}/utils/src/protobuf"
+    PROTOC_OUT_DIR "${PROTO_BINARY_DIR}")
+
 if (CLIENT)
   add_definitions("-D CLIENT")
 
@@ -107,7 +123,7 @@ else()
 
   add_executable(vdms src/vdms.cc)
 
-  target_link_libraries(vdms dms vdms_protobuf vcl tiledb faiss flinng jsoncpp ${OpenCV_LIBS} ${AWSSDK_LINK_LIBRARIES})
+  target_link_libraries(vdms dms vdms_protobuf vcl tiledb faiss flinng jsoncpp ${OpenCV_LIBS} ${AWSSDK_LINK_LIBRARIES} ${Protobuf_LIBRARIES} GTest::gtest GTest::gtest_main)
 endif ()
 
 message("Coverage:" ${CODE_COVERAGE})

client/python/setup.py

@@ -9,7 +9,7 @@
     author="Chaunté W. Lacewell",
     author_email="[email protected]",
     description="VDMS Client Module",
-    install_requires=["protobuf==4.24.2"],
+    install_requires=["protobuf==6.31.0"],
     long_description=long_description,
     long_description_content_type="text/markdown",
     url="https://github.com/IntelLabs/vdms",

include/vcl/GRPCEntityClient.h

@@ -0,0 +1,54 @@
+#ifndef GRPC_ENTITY_CLIENT_H
+#define GRPC_ENTITY_CLIENT_H
+
+#include <grpcpp/alarm.h>
+#include <grpcpp/grpcpp.h>
+
+#include <condition_variable>
+#include <map>
+#include <mutex>
+#include <string>
+#include <thread>
+
+#include "Exception.h"
+#include "entity.grpc.pb.h"
+
+class GRPCEntityClient {
+ public:
+  explicit GRPCEntityClient(std::string url);
+
+  void ProcessEntities(const std::map<std::string, std::string>& input_paths,
+                       const std::map<std::string, std::string>& output_paths,
+                       const std::map<std::string, std::string>& input_metadata,
+                       std::map<std::string, std::string>& output_metadata,
+                       bool& success);
+
+ private:
+  struct AsyncCall;
+
+  void InitStub();
+  void SendRequest(const std::string& entity_id, const std::string& input_path);
+  bool ReadFile(const std::string& path, std::string& out);
+  bool WriteFile(const std::string& path, const std::string& data);
+  void WaitForSlot();
+  void HandleRpcs();
+
+  std::unique_ptr<entity::Operator::Stub> stub_;
+  std::string url_;
+  grpc::CompletionQueue cq_;
+  std::thread worker_;
+  bool worker_started_ = false;
+
+  const std::map<std::string, std::string>* output_paths_;
+  const std::map<std::string, std::string>* input_metadata_;
+  std::map<std::string, std::string>* output_metadata_;
+  bool* success_;
+
+  std::mutex mutex_;
+  std::condition_variable cond_;
+  std::condition_variable all_done_;
+  int in_flight_ = 0;
+  int max_concurrent_tasks_;
+};
+
+#endif  // GRPC_ENTITY_CLIENT_H

include/vcl/Image.h

@@ -36,6 +36,7 @@
 #include <curl/curl.h>
 #include <jsoncpp/json/reader.h>
 #include <jsoncpp/json/value.h>
+#include <jsoncpp/json/writer.h>
 #include <stdio.h>
 
 #include <fstream>
@@ -48,6 +49,7 @@
 #include <zmq.hpp>
 
 #include "Exception.h"
+#include "GRPCEntityClient.h"
 #include "RemoteConnection.h"
 #include "TDBImage.h"
 #include "VDMSConfigHelper.h"

include/vcl/Video.h

@@ -32,6 +32,8 @@
 #pragma once
 
 #include <jsoncpp/json/reader.h>
+#include <jsoncpp/json/value.h>
+#include <jsoncpp/json/writer.h>
 
 #include <fstream>
 #include <list>
@@ -49,7 +51,6 @@
 #include "timers/TimerMap.h"
 #include "utils.h"
 #include "vcl/Image.h"
-#include "zip.h"
 
 namespace VCL {
 

remote_function/entity.proto

@@ -0,0 +1,12 @@
+syntax = "proto3";
+
+package entity;
+
+service Operator {
+  rpc Operate (Entity) returns (Entity);
+}
+
+message Entity {
+  bytes entity = 1;
+  bytes options = 2;
+}

remote_function/functions/caption.py

@@ -1,38 +1,32 @@
+import imageio.v3 as iio
+import skvideo.io
 import cv2
-import uuid
 import os
+import uuid
 
 
-def run(ipfilename, format, options, tmp_dir_path=""):
-    opfilename = os.path.join(
-        tmp_dir_path, "tmpfile" + uuid.uuid1().hex + "." + str(format)
-    )
-
-    vc = cv2.VideoCapture(ipfilename)
-    frame_width = int(vc.get(cv2.CAP_PROP_FRAME_WIDTH))
-    frame_height = int(vc.get(cv2.CAP_PROP_FRAME_HEIGHT))
-    video_fps = vc.get(cv2.CAP_PROP_FPS)
-
-    video = cv2.VideoWriter(
-        opfilename,
-        cv2.VideoWriter_fourcc(*"mp4v"),
-        video_fps,
-        (frame_width, frame_height),
-    )
-
-    while True:
-        (grabbed, frame) = vc.read()
-        if not grabbed:
-            print("[INFO] no frame read from stream - exiting")
-            break
+def run(entity, options, tmp_dir_path=""):
+    fname = os.path.join(tmp_dir_path, "tmpfile" + uuid.uuid1().hex + ".mp4")
 
-        label = options["text"]
+    label = options["text"]
+    video = skvideo.io.FFmpegWriter(fname)
+    for frame in iio.imiter(entity, format_hint=".mp4"):
         cv2.putText(
             frame, label, (10, 25), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (255, 255, 255), 2
         )
+        video.writeFrame(frame)
+
+    video.close()
+
+    ebytes = ""
+    with open(fname, "rb") as f:
+        ebytes = f.read()
+
+    # with open('bytefile.mp4', "wb") as out_file:
+    #     out_file.write(ebytes)
+
+    os.remove(fname)
 
-        video.write(frame)
-    vc.release()
-    video.release()
+    rdict = {"metadata": "None"}
 
-    return opfilename, None
+    return ebytes, rdict

remote_function/functions/facedetect.py

@@ -1,5 +1,6 @@
 import cv2
 import os
+import numpy as np
 
 # Get the real directory where this Python file is
 currentDir = os.path.realpath(os.path.dirname(__file__))
@@ -20,19 +21,23 @@
 )
 
 
-def run(ipfilename, format, options, tmp_dir_path=""):
+def run(entity, options):
     global face_cascade
 
-    if not os.path.exists(ipfilename):
-        raise Exception(
-            f"Facedetect error: File ipfilename: {ipfilename} does not exist"
-        )
+    image_array = np.frombuffer(entity, dtype=np.uint8)
 
-    img = cv2.imread(ipfilename)
+    img = cv2.imdecode(image_array, cv2.IMREAD_COLOR)
     gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
     faces = face_cascade.detectMultiScale(gray, 1.1, 4)
 
     for x, y, w, h in faces:
         cv2.rectangle(img, (x, y), (x + w, y + h), (255, 0, 0), 2)
 
-    return img, None
+    success, encoded_img = cv2.imencode(".jpg", img)
+    if not success:
+        raise ValueError("Failed to encode image.")
+    ebytes = encoded_img.tobytes()
+
+    rdict = {"metadata": "None"}
+
+    return ebytes, rdict