Protobuf send (#10)

* feat: modify osd mapping util for handoff conn

* feat: add handoff_in accept in main

* feat: multiplex for handoff_in/out and s3_http on epoll.event.data.u32

* docs(handoff): more implementation details about handoff requests

* fix compile

* fix: delete tlse related macros

* factor handoff code to own source file

* implement socket sock and client serialization

- fix control plane loop to use epoll event data ptr
- use first element of structures to indicate event type

* link libforward-tc

* feat: handoff out

* feat: handoff in

* fix compile

* switch to use val for rados ctx

* remove aio

* set first read to 512

* refactor

* bugfix

* fix to_migrate

* add more prints

* fetch client MAC address and add to migration info

* implement soruce IP modification on fake server

* temp use async for GET

* modify aio

* fake server reply HTTP OK and data

* handoff out reconnect max try

* build options to build for async io and migration

* ktls serialize and deserialize

* incoming traffic block with before deserialize

* add redirection rule

* fix missing http reset after migration

* implement migration option switch

* handoff back

* handoff back and back-forward working

---------

Co-authored-by: steven <steven.chien@ed.ac.uk>

Author: breakertt

.gitignore

@@ -54,3 +54,6 @@ core.*
 vgcore.*
 
 build
+
+proto/socket_serialize.pb-c.h
+proto/socket_serialize.pb-c.c

CMakeLists.txt

@@ -2,25 +2,40 @@ cmake_minimum_required(VERSION 3.22)
 project(xo-server)
 enable_language(C)
 
+include(ExternalProject)
 include(FetchContent)
 
 include(${CMAKE_SOURCE_DIR}/cmake/modules/Findlibrados2.cmake)
 include(${CMAKE_SOURCE_DIR}/cmake/modules/FindProtobufC.cmake)
 list(APPEND CMAKE_MODULE_PATH "${CMAKE_SOURCE_DIR}/cmake/modules")
 
-#option(BUILD_SHARED_LIBS "Build using shared libraries" OFF)
+option(USE_ASYNC_WRITE "Build using rados aio for write" OFF)
+option(USE_ASYNC_READ "Build using rados aio for read" OFF)
+option(USE_MIGRATION "Build migration support" OFF)
+
 FetchContent_Declare(llhttp
-  URL "https://github.com/nodejs/llhttp/archive/refs/tags/release/v8.1.0.tar.gz")
+    URL "https://github.com/nodejs/llhttp/archive/refs/tags/release/v8.1.0.tar.gz")
 
 set(BUILD_SHARED_LIBS OFF CACHE INTERNAL "")
 set(BUILD_STATIC_LIBS ON CACHE INTERNAL "")
 FetchContent_MakeAvailable(llhttp)
 
+# add libforward source
+ExternalProject_Add(libforward-tc
+    GIT_REPOSITORY https://github.com/uoenoplab/libforward-tc.git
+    SOURCE_DIR libs/libforward-tc
+    UPDATE_DISCONNECTED true
+    GIT_TAG main
+    #BUILD_IN_SOURCE ON
+    #CONFIGURE_COMMAND ./configure
+    #BUILD_COMMAND make -j 32
+    INSTALL_COMMAND "")
+
 ## Link with the llhttp_static target
 #target_link_libraries(${EXAMPLE_PROJECT_NAME} ${PROJECT_LIBRARIES} llhttp_static ${PROJECT_NAME})
 
 if (NOT CMAKE_BUILD_TYPE)
-  set(CMAKE_BUILD_TYPE "Debug")
+    set(CMAKE_BUILD_TYPE "Debug")
 endif()
 
 if (POLICY CMP0074)
@@ -30,7 +45,7 @@ endif()
 set(C_SRC
     src/http_client.c
     src/object_store.c
-    src/conn_migration.c
+    src/handoff.c
     src/md5.c
     src/util.c
 )
@@ -55,48 +70,81 @@ find_program(Protobuf_PROTOC-C_EXECUTABLE
 mark_as_advanced(Protobuf_PROTOC-C_EXECUTABLE)
 PROTOBUF_GENERATE_C(PROTO_SRC PROTO_HEAD proto/socket_serialize.proto)
 
-# build benchmark library
+# build http server library
 add_library(http_server STATIC
     ${PROTO_SRC}
     ${C_SRC})
+
+add_dependencies(http_server libforward-tc)
+
+ExternalProject_Get_property(libforward-tc SOURCE_DIR)
 target_include_directories(http_server PRIVATE
     include
     ${CMAKE_CURRENT_BINARY_DIR} # for proto
+    ${SOURCE_DIR}/include
     ${URIPARSER_INCLUDE_DIR}
     ${llhttp_SOURCE_DIR}/include
     ${LIBXML2_INCLUDE_DIR}
     ${LIBRADOS2_INCLUDE_DIRS})
-target_link_libraries(http_server PRIVATE
+
+ExternalProject_Get_property(libforward-tc BINARY_DIR)
+target_link_directories(http_server PUBLIC
+    ${BINARY_DIR})
+
+target_link_libraries(http_server PUBLIC
     llhttp_static
     uriparser::uriparser
     LibXml2::LibXml2
     OpenSSL::Crypto
+    forward-tc
+    ${PROTOBUF-C_LIBRARIES}
     ${LIBRADOS2_LIBRARIES})
+
 set_target_properties(http_server PROPERTIES
-        POSITION_INDEPENDENT_CODE ON
-        C_STANDARD 11)
+    POSITION_INDEPENDENT_CODE ON
+    C_STANDARD 11)
+
+if (USE_ASYNC_WRITE)
+	list(APPEND HTTP_SERVER_DEF -DASYNC_WRITE)
+endif()
+if (USE_ASYNC_READ)
+	list(APPEND HTTP_SERVER_DEF -DASYNC_READ)
+endif()
+if (USE_MIGRATION)
+	list(APPEND HTTP_SERVER_DEF -DUSE_MIGRATION)
+endif()
+
+target_compile_definitions(http_server PUBLIC ${HTTP_SERVER_DEF})
 target_compile_options(http_server PRIVATE
-	-fstack-protector-all -Wall -Wextra -ggdb -O3 -g)
+    -fstack-protector-all -Wall -Wextra -ggdb -O3 -g)
 #-fsanitize=undefined,address -fno-omit-frame-pointer -fstack-protector -Wall -Wextra -ggdb -O3 -g)
 target_link_options(http_server PRIVATE
-	-fstack-protector-all -Wall -Wextra -ggdb -O3 -g)
+    -fstack-protector-all -Wall -Wextra -ggdb -O3 -g)
 #-fsanitize=undefined,address -fno-omit-frame-pointer -fstack-protector -Wall -Wextra -ggdb -O3 -g)
 
 # build driver application
 add_executable(server.out src/server.c src/tls.c)
+
+add_dependencies(server.out libforward-tc http_server)
+
 target_include_directories(server.out PUBLIC
+    ${SOURCE_DIR}/include
     ${llhttp_SOURCE_DIR}/include
     ${INIH_INCLUDE_DIRS}
     ${LIBXML2_INCLUDE_DIR}
     include)
+
 target_link_libraries(server.out PUBLIC
+    llhttp_static
     http_server
     ${INIH_LIBRARIES}
-    ${PROTOBUF-C_LIBRARIES}
+    #efence
     OpenSSL::Crypto
     OpenSSL::SSL)
+
 set_target_properties(server.out PROPERTIES
     C_STANDARD 11)
+
 target_compile_options(server.out PRIVATE
     ${INIH_CFLAGS_OTHER}
     -fstack-protector-all -O3 -Wall -Wextra -g)
@@ -113,6 +161,9 @@ target_link_options(server.out PRIVATE
 file(COPY ${CMAKE_SOURCE_DIR}/assets DESTINATION ${CMAKE_BINARY_DIR})
 
 message(STATUS "Building xo server binary")
+message(STATUS "Build with rados aio for write: " ${USE_ASYNC_WRITE})
+message(STATUS "Build with rados aio for read: " ${USE_ASYNC_READ})
+message(STATUS "Build with connection migration support: " ${USE_MIGRATION})
 
 # uninstall target
 #if(NOT TARGET uninstall)

include/conn_migration.h

@@ -0,0 +1,84 @@
+#ifndef __HANDOFF_H__
+#define __HANDOFF_H__
+
+#include "forward.h"
+#include "ebpf_forward.h"
+
+#include "http_client.h"
+
+#ifndef TCPOPT_MSS
+#define TCPOPT_MSS 2
+#endif
+
+#ifndef TCPOPT_WINDOW
+#define TCPOPT_WINDOW 3
+#endif
+
+#ifndef TCPOPT_SACK_PERM
+#define TCPOPT_SACK_PERM 4
+#endif
+
+#ifndef TCPOPT_TIMESTAMP
+#define TCPOPT_TIMESTAMP 8
+#endif
+
+enum {
+        HANDOFF_REQUEST,
+        HANDOFF_BACK_REQUEST,
+        HANDOFF_DONE
+};
+
+struct handoff_in {
+	uint32_t epoll_data_u32;
+	int epoll_fd;
+	int fd;
+        int osd_arr_index;
+        int thread_id; // just for print stuff
+        uint8_t *recv_protobuf;
+        uint32_t recv_protobuf_len;
+        uint32_t recv_protobuf_received;
+        uint8_t *send_protobuf;
+        uint32_t send_protobuf_len; // include header uint32 size
+        uint32_t send_protobuf_sent;
+	rados_ioctx_t data_io_ctx;
+	rados_ioctx_t bucket_io_ctx;
+        struct http_client *client_to_handoff_again;
+};
+
+struct handoff_out_req;
+
+struct handoff_out_queue {
+    struct handoff_out_req* front;
+    struct handoff_out_req* rear;
+};
+
+#define MAX_HANDOFF_OUT_RECONNECT 10
+
+struct handoff_out {
+	uint32_t epoll_data_u32;
+	int epoll_fd;
+	int fd;
+        bool is_fd_in_epoll;
+        int reconnect_count;
+        int osd_arr_index;
+        int thread_id;  // just for print stuff
+        struct handoff_out_queue *queue;
+        // handoff out request currently sending out, deququed from queue
+        struct http_client *client;
+        uint8_t *recv_protobuf;
+        uint32_t recv_protobuf_len;
+        uint32_t recv_protobuf_received;
+};
+
+void handoff_out_connect(struct handoff_out *out_ctx);
+void handoff_out_reconnect(struct handoff_out *out_ctx);
+void handoff_out_issue(int epoll_fd, uint32_t epoll_data_u32, struct http_client *client,
+	struct handoff_out *out_ctx, int osd_arr_index, int thread_id, bool serialize, bool urgent);
+void handoff_out_send(struct handoff_out *out_ctx);
+void handoff_out_recv(struct handoff_out *out_ctx);
+
+void handoff_in_disconnect(struct handoff_in *in_ctx);
+void handoff_in_recv(struct handoff_in *in_ctx);
+void handoff_in_send(struct handoff_in *in_ctx, struct http_client **client_to_handoff_again);
+
+#endif

include/handoff.h

@@ -24,7 +24,10 @@
 
 enum http_expect { CONTINUE, NONE };
 
-extern size_t BUF_SIZE;
+extern const size_t BUF_SIZE;
+#ifdef USE_MIGRATION
+extern bool enable_migration;
+#endif
 
 typedef struct ssl_st SSL;
 typedef struct bio_st BIO;
@@ -42,17 +45,16 @@ static char *HTTP_CONTINUE_HDR = (char *)"HTTP/1.1 100 CONTINUE\r\n\r\n";
 static char *AMZ_REQUEST_ID = (char*)"tx000009a75d393f1564ec2-0065202454-3771-default";
 
 struct http_client {
-	int epoll_fd;
+	uint32_t epoll_data_u32;
+	// always set epoll.data.u32 to this value
 	int fd;
+	int epoll_fd;
 
 	int prval;
 	rados_xattrs_iter_t iter;
-
-	rados_write_op_t write_op;
+	rados_completion_t comp;
 	rados_read_op_t read_op;
-
-	rados_completion_t aio_completion;
-	rados_completion_t aio_head_read_completion;
+	rados_read_op_t write_op;
 
 	ssize_t data_payload_sent;
 	ssize_t data_payload_size;
@@ -86,8 +88,11 @@ struct http_client {
 	//char **header_values;
 	size_t num_fields;
 
+	pthread_mutex_t mutex;
 	rados_ioctx_t *bucket_io_ctx;
 	rados_ioctx_t *data_io_ctx;
+	rados_completion_t aio_completion;
+	rados_completion_t aio_head_read_completion;
 
 	size_t object_size;
 	size_t object_offset;
@@ -123,6 +128,16 @@ struct http_client {
 		unsigned char client_traffic_secret[48];
 		unsigned char server_traffic_secret[48];
 	} tls;
+
+	// handoff
+	int to_migrate;
+	int from_migrate;
+	int acting_primary_osd_id;
+	uint8_t client_mac[6];
+
+	uint8_t *proto_buf;
+	uint32_t proto_buf_len; // include the uint32 header size
+	uint32_t proto_buf_sent;
 };
 
 struct http_client *create_http_client(int epoll_fd, int fd);
@@ -139,9 +154,9 @@ void free_http_client(struct http_client *client);
 //static int on_message_complete_cb(llhttp_t* parser);
 //static int on_reset_cb(llhttp_t *parser);
 
-void send_client_data(struct http_client *client);
-void send_response(struct http_client *client);
 void aio_ack_callback(rados_completion_t comp, void *arg);
 void aio_commit_callback(rados_completion_t comp, void *arg);
+void send_client_data(struct http_client *client);
+void send_response(struct http_client *client);
 
 #endif

include/http_client.h

@@ -1,8 +1,43 @@
 #ifndef __OSD_MAPPING_H__
 #define __OSD_MAPPING_H__
 
-extern char *osd_ips[65536];
-extern long my_osd_id;
-extern char *my_ip_address;
+#define MAX_OSDS 64
+
+extern char *osd_addr_strs[MAX_OSDS];
+extern struct sockaddr_in osd_addrs[MAX_OSDS];
+extern int osd_ids[MAX_OSDS];
+extern int osd_id_to_arr_index[];
+extern int num_osds;
+extern int num_peers;
+
+extern uint8_t my_mac[6];
+
+// FUTURE TODO: add a data structure to optimize this linear search?
+static inline int get_arr_index_from_osd_id(int osd_id)
+{
+    for (int i = 0; i < num_osds; i++)
+    {
+        if (osd_ids[i] == osd_id)
+            return i;
+    }
+    fprintf(stderr, "%s: cant find osd_id %d in array\n",
+        __func__, osd_id);
+    return -1;
+}
+
+static inline struct sockaddr_in get_my_osd_addr()
+{
+    return osd_addrs[num_osds - 1];
+}
+
+static inline char *get_my_osd_addr_str()
+{
+    return osd_addr_strs[num_osds - 1];
+}
+
+static inline int get_my_osd_id()
+{
+    return osd_ids[num_osds - 1];
+}
 
 #endif

include/osd_mapping.h

@@ -1,6 +1,8 @@
 #ifndef __UTIL_H__
 #define __UTIL_H__
 
+void print_mac_address(unsigned char *mac);
+void hexdump(const char *title, void *buf, size_t len);
 double elapsed_time(struct timespec a, struct timespec b);
 void get_datetime_str(char *buf, size_t length);
 void convertToISODateTime(const char* inputDateTime, char* outputISODateTime);