d/ppd
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

master slave mode working

Browse Source
This commit is contained in:
quackerd 2023-03-14 14:30:03 +01:00
parent e97dd8ee9e
commit 227d65a235
6 changed files with 30 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
CMakeLists.txt
View File

@ -6,8 +6,8 @@ find_package(PkgConfig REQUIRED)
find_package(OpenSSL REQUIRED) find_package(OpenSSL REQUIRED)
pkg_check_modules(bsock REQUIRED bsock) pkg_check_modules(bsock REQUIRED bsock)
set(CXXFLAGS -Wall -Wextra -Werror -std=c++20 -O3 -g) set(CXXFLAGS -Wall -Wextra -Werror -std=c++17 -Og -g)
set(CFLAGS -Wall -Wextra -Werror -std=c2x -O3 -g) set(CFLAGS -Wall -Wextra -Werror -std=c2x -Og -g)
add_library(common OBJECT ${CMAKE_CURRENT_SOURCE_DIR}/common/io.cc) add_library(common OBJECT ${CMAKE_CURRENT_SOURCE_DIR}/common/io.cc)
target_link_directories(common PRIVATE ${bsock_LIBRARY_DIRS}) target_link_directories(common PRIVATE ${bsock_LIBRARY_DIRS})

2
dsmbr/dmsg.cc
View File

@ -13,7 +13,7 @@ dsmbr_send_ctrl_msg(int sockfd, struct dsmbr_ctrl_msg * msg)
msg->data[i] = htobe64(msg->data[i]); msg->data[i] = htobe64(msg->data[i]);
} }
return ppd_writebuf(sockfd, &msg, sizeof(struct dsmbr_ctrl_msg)); return ppd_writebuf(sockfd, msg, sizeof(struct dsmbr_ctrl_msg));
} }
int int

2
dsmbr/dmsg.hh
View File

@ -10,7 +10,7 @@ static constexpr int CTRL_STOP = 0x3456;
static constexpr int CTRL_STAT = 0x4567; static constexpr int CTRL_STAT = 0x4567;
static constexpr int DSMBR_MSG_MAX_DATA = 16; static constexpr int DSMBR_MSG_MAX_DATA = 16;
struct dsmbr_ctrl_msg { struct __attribute__((packed)) dsmbr_ctrl_msg {
uint32_t code; uint32_t code;
uint64_t data[DSMBR_MSG_MAX_DATA]; uint64_t data[DSMBR_MSG_MAX_DATA];
}; };

35
dsmbr/dsmbr.cc
View File

@ -46,7 +46,7 @@ static constexpr int MAX_MOD_ARGS = 32;
static constexpr int MAX_MOD_ARG_LEN = 128; static constexpr int MAX_MOD_ARG_LEN = 128;
static constexpr int MAX_SLAVES = 32; static constexpr int MAX_SLAVES = 32;
static constexpr int NEVENT = 64; static constexpr int NEVENT = 64;
static constexpr int CTRL_PORT = 15367; static constexpr int CTRL_PORT = 15545;
static constexpr int CTRL_TIMEOUT = 3; static constexpr int CTRL_TIMEOUT = 3;
static constexpr int CTRL_BACKLOG = 4096; static constexpr int CTRL_BACKLOG = 4096;
static constexpr int BSOCK_BUF_SZ = 4096; static constexpr int BSOCK_BUF_SZ = 4096;
@ -519,21 +519,21 @@ dsmbr_slave_ctrl_sock_create(void)
int fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); int fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if (fd < 0) { if (fd < 0) {
E("socket() returned %d", errno); E("socket() returned %d\n", errno);
} }
if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &enable, sizeof(enable)) < 0) { if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &enable, sizeof(enable)) < 0) {
E("setsockopt() NODELAY %d", errno); E("setsockopt() NODELAY %d\n", errno);
} }
status = bind(fd, (struct sockaddr *)&server_addr, sizeof(server_addr)); status = bind(fd, (struct sockaddr *)&server_addr, sizeof(server_addr));
if (status < 0) { if (status < 0) {
E("bind() returned %d", errno); E("bind() returned %d\n", errno);
} }
status = listen(fd, CTRL_BACKLOG); status = listen(fd, CTRL_BACKLOG);
if (status < 0) { if (status < 0) {
E("listen() returned %d", errno); E("listen() returned %d\n", errno);
} }
return fd; return fd;
@ -567,8 +567,6 @@ dsmbr_slave_accept_master(int slave_ctrl_sock)
int conn_fd; int conn_fd;
struct sockaddr addr; struct sockaddr addr;
socklen_t addrlen; socklen_t addrlen;
char ip[INET_ADDRSTRLEN];
struct in_addr inaddr = ((struct sockaddr_in *)&addr)->sin_addr;
ret = accept(slave_ctrl_sock, &addr, &addrlen); ret = accept(slave_ctrl_sock, &addr, &addrlen);
if (ret < 0) { if (ret < 0) {
@ -576,11 +574,7 @@ dsmbr_slave_accept_master(int slave_ctrl_sock)
} }
conn_fd = ret; conn_fd = ret;
if (inet_ntop(AF_INET, &inaddr, ip, INET_ADDRSTRLEN) == NULL) { V("Accepted master connection %d.\n", conn_fd);
E("inet_ntop failed with %d\n", errno);
}
V("Accepted master connection %d from %s.", conn_fd, ip);
return conn_fd; return conn_fd;
} }
@ -677,7 +671,7 @@ dsmbr_getopt(int argc, char *argv[])
} }
static void static void
dsmbr_send_ctrl_code(int sock, int code) dsmbr_send_ctrl_code(int sock, uint32_t code)
{ {
int ret; int ret;
struct dsmbr_ctrl_msg msg; struct dsmbr_ctrl_msg msg;
@ -690,7 +684,7 @@ dsmbr_send_ctrl_code(int sock, int code)
} }
static void static void
dsmbr_wait_ctrl_code(int sock, int code) dsmbr_wait_ctrl_code(int sock, uint32_t code)
{ {
int ret; int ret;
struct dsmbr_ctrl_msg msg; struct dsmbr_ctrl_msg msg;
@ -698,8 +692,8 @@ dsmbr_wait_ctrl_code(int sock, int code)
if (ret != 0) { if (ret != 0) {
E("Failed to recv ctrl message %d\n", errno); E("Failed to recv ctrl message %d\n", errno);
} }
if (msg.code != CTRL_SYNC) { if (msg.code != code) {
E("Unexpected message code %d, expected %d\n", msg.code, code); E("Unexpected message code 0x%x, expected 0x%x\n", msg.code, code);
} }
return; return;
} }
@ -771,13 +765,14 @@ main(int argc, char *argv[])
V("S: waiting for master connection...\n"); V("S: waiting for master connection...\n");
master_ctrl_conn = dsmbr_slave_accept_master(slave_ctrl_sock); master_ctrl_conn = dsmbr_slave_accept_master(slave_ctrl_sock);
// add to kqueue // add to kqueue
V("S: waiting for master SYNC %d...\n", master_ctrl_conn);
dsmbr_wait_ctrl_code(master_ctrl_conn, CTRL_SYNC);
struct kevent kev; struct kevent kev;
EV_SET(&kev, master_ctrl_conn, EVFILT_READ, EV_ADD, 0, 0, NULL); EV_SET(&kev, master_ctrl_conn, EVFILT_READ, EV_ADD, 0, 0, NULL);
if (kevent(kqfd, &kev, 1, NULL, 0, NULL) == -1) { if (kevent(kqfd, &kev, 1, NULL, 0, NULL) == -1) {
E("kevent() failed: %d\n", errno); E("kevent() failed: %d\n", errno);
} }
V("S: waiting for master SYNC...\n");
dsmbr_wait_ctrl_code(master_ctrl_conn, CTRL_SYNC);
} }
// here slaves and master are on the same page */ // here slaves and master are on the same page */
@ -907,14 +902,14 @@ main(int argc, char *argv[])
ctrl_msg.data[0] = total_req; ctrl_msg.data[0] = total_req;
ctrl_msg.data[1] = end_ts - start_ts; ctrl_msg.data[1] = end_ts - start_ts;
ctrl_msg.data[2] = total_send; ctrl_msg.data[2] = total_send;
ctrl_msg.data[4] = total_recv; ctrl_msg.data[3] = total_recv;
if (dsmbr_send_ctrl_msg(master_ctrl_conn, &ctrl_msg) != 0) { if (dsmbr_send_ctrl_msg(master_ctrl_conn, &ctrl_msg) != 0) {
E("dsmbr_send_ctrl_msg failed with %d.\n", errno); E("dsmbr_send_ctrl_msg failed with %d.\n", errno);
} }
} }
V("Saving results to %s ...\n", options.output_file);
if (!options.slave_mode) { if (!options.slave_mode) {
V("Saving results to %s ...\n", options.output_file);
for (int i = 0; i < options.num_slaves; i++) { for (int i = 0; i < options.num_slaves; i++) {
fprintf(resp_fp_csv, "S,%lu,%lu,%lu,%lu\n", slaves[i].reqs, slaves[i].time, fprintf(resp_fp_csv, "S,%lu,%lu,%lu,%lu\n", slaves[i].reqs, slaves[i].time,
slaves[i].send_sz, slaves[i].recv_sz); slaves[i].send_sz, slaves[i].recv_sz);

9
include/msg.hh
View File

@ -1,15 +1,16 @@
#pragma once #pragma once
#include <cstdint> #include <cstdint>
#include <cstddef>
// Max 16MB per message // Max 16MB per message
struct ppd_msg { struct __attribute__((packed)) ppd_msg {
uint32_t size; uint32_t size;
char data[0]; char data[0];
} __attribute__((packed)); };
static constexpr int PPD_MSG_MAX_SZ = (1024 * 16 * 1024); static constexpr int PPD_MSG_MAX_SZ = 1024 * 16 * 1024;
static constexpr int PPD_MSG_HDR_SZ = offsetof(struct ppd_msg, data); static constexpr int PPD_MSG_HDR_SZ = offsetof(ppd_msg, data);
static_assert(PPD_MSG_HDR_SZ == sizeof(struct ppd_msg)); static_assert(PPD_MSG_HDR_SZ == sizeof(struct ppd_msg));
static constexpr int PPD_MSG_MAX_DATA_SZ = (PPD_MSG_MAX_SZ - PPD_MSG_HDR_SZ); static constexpr int PPD_MSG_MAX_DATA_SZ = (PPD_MSG_MAX_SZ - PPD_MSG_HDR_SZ);

12
ppd/ppd.cc
View File

@ -190,29 +190,29 @@ listen_socket_create(std::vector<int> *socks)
for (int i = 0; i < 1; i++) { for (int i = 0; i < 1; i++) {
int fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); int fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if (fd < 0) { if (fd < 0) {
E("socket() returned %d", errno); E("socket() returned %d\n", errno);
} }
if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(enable)) < 0) { if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(enable)) < 0) {
E("setsockopt() reuseaddr %d", errno); E("setsockopt() reuseaddr %d\n", errno);
} }
if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &enable, sizeof(enable)) < 0) { if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &enable, sizeof(enable)) < 0) {
E("setsockopt() reuseport %d", errno); E("setsockopt() reuseport %d\n", errno);
} }
if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &enable, sizeof(enable)) < 0) { if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &enable, sizeof(enable)) < 0) {
E("setsockopt() NODELAY %d", errno); E("setsockopt() NODELAY %d\n", errno);
} }
status = bind(fd, (struct sockaddr *)&server_addr, sizeof(server_addr)); status = bind(fd, (struct sockaddr *)&server_addr, sizeof(server_addr));
if (status < 0) { if (status < 0) {
E("bind() returned %d", errno); E("bind() returned %d\n", errno);
} }
status = listen(fd, SOCK_BACKLOG); status = listen(fd, SOCK_BACKLOG);
if (status < 0) { if (status < 0) {
E("listen() returned %d", errno); E("listen() returned %d\n", errno);
} }
socks->push_back(fd); socks->push_back(fd);