io baseline

src/CMakeLists.txt

@@ -7,6 +7,8 @@ set(LIB_SRC
   ${SOURCE_DIR}/kad_network.cpp
   ${SOURCE_DIR}/kad_node.cpp
   ${SOURCE_DIR}/kad_routable.cpp
+  ${SOURCE_DIR}/io_client.cpp
+  ${SOURCE_DIR}/io_server.cpp
   ${SOURCE_DIR}/shell.cpp
 
   CACHE
@@ -26,6 +28,7 @@ configure_file(${SOURCE_DIR}/config.in ${GENERATE_DIR}/config.h @ONLY)
 find_package(OpenSSL          REQUIRED)
 find_package(Readline         REQUIRED)
 find_package(JsonRpcCppClient REQUIRED)
+find_package(MHD              REQUIRED)
 find_package(NTTEC            REQUIRED)
 
 get_property(NTTEC_INCLUDE_DIRS
@@ -37,6 +40,7 @@ set(OBJECT_SYS_INCLUDES
   ${GENERATE_DIR}
   ${OPENSSL_INCLUDE_DIR}
   ${JsonRpcCppClient_INCLUDE_DIRS}
+  ${MHD_INCLUDE_DIRS}
   ${Readline_INCLUDE_DIRS}
   ${NTTEC_INCLUDE_DIRS}
 )
@@ -75,6 +79,7 @@ foreach(lib ${SHARED_LIB} ${STATIC_LIB})
     ${OPENSSL_CRYPTO_LIBRARY}
     ${Readline_LIBRARIES}
     ${JsonRpcCppClient_LIBRARIES}
+    ${MHD_LIBRARIES}
     NTTEC::static
   )
 endforeach()

src/cmds.cpp

@@ -8,6 +8,7 @@
 #include "kad_network.h"
 #include "kad_node.h"
 #include "kad_routable.h"
+#include "io_client.h"
 #include "shell.h"
 #include "utils.h"
 
@@ -337,6 +338,33 @@ static int cmd_get_bytes(Shell* shell, int argc, char** argv)
     return SHELL_CONT;
 }
 
+static int cmd_put(Shell * /* shell */, int argc, char **argv)
+{
+  if (argc != 4)
+    {
+      fprintf(stderr, "usage: put file n_data n_parities\n");
+      return SHELL_CONT;
+    }
+
+  do_put(argv[1], atoi(argv[2]), atoi(argv[3]));
+
+  return SHELL_CONT;
+}
+
+static int cmd_io_server_info(Shell *shell, int, char **)
+{
+  auto* node = static_cast<Node*>(shell->get_handle2());
+
+  if (nullptr == node) {
+    std::cerr << "shall jump to a node first\n";
+    return SHELL_CONT;
+  }
+
+  std::cout << "port: " << node->io_server->get_port() << "\n";
+
+  return SHELL_CONT;
+}
+
 struct cmd_def quit_cmd = {"quit", "quit program", cmd_quit};
 struct cmd_def help_cmd = {"help", "help", cmd_help};
 struct cmd_def jump_cmd = {"jump", "jump to a node", cmd_jump};
@@ -353,6 +381,7 @@ struct cmd_def find_nearest_cmd = {"find_nearest",
                                    "find nearest nodes to",
                                    cmd_find_nearest};
 struct cmd_def verbose_cmd = {"verbose", "set verbosity level", cmd_verbose};
+struct cmd_def put_cmd = {"put", "put a file", cmd_put};
 struct cmd_def save_cmd = {"save", "save the network to file", cmd_save};
 struct cmd_def xor_cmd = {"xor", "xor between 2 bignums", cmd_xor};
 struct cmd_def bit_length_cmd = {"bit_length",
@@ -371,11 +400,15 @@ struct cmd_def put_bytes_cmd = {"put_bytes",
 struct cmd_def get_bytes_cmd = {"get_bytes",
                                 "get N bytes from storage",
                                 cmd_get_bytes};
-
+struct cmd_def io_server_info_cmd = {"io_server_info",
+                                     "get IO server info",
+                                     cmd_io_server_info};
+
 struct cmd_def* cmd_defs[] = {
     &bit_length_cmd,
     &buy_storage_cmd,
     &cheat_lookup_cmd,
+    &put_cmd,
     &find_nearest_cmd,
     &get_bytes_cmd,
     &graphviz_cmd,
@@ -390,6 +423,7 @@ struct cmd_def* cmd_defs[] = {
     &show_cmd,
     &verbose_cmd,
     &xor_cmd,
+    &io_server_info_cmd,
     nullptr,
 };
 

src/io_client.cpp

@@ -0,0 +1,124 @@
+
+#include <cstdio>
+#include <cstdlib>
+#include <streambuf>
+
+#include <sys/mman.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <nttec/nttec.h>
+
+#include "io_client.h"
+
+template class nttec::fec::RsFnt<uint32_t>;
+
+int vflag = 1;
+
+struct membuf : std::streambuf {
+  membuf(char const *base, size_t size) {
+    char *p(const_cast<char *>(base));
+    this->setg(p, p, p + size);
+  }
+};
+
+struct imemstream : virtual membuf, std::istream {
+  imemstream(char const *base, size_t size)
+      : membuf(base, size), std::istream(static_cast<std::streambuf *>(this)) {}
+};
+
+/**
+ * Create coding files for the specified file
+ *
+ * This function:
+ * 1) compute the ideal mmap size
+ * 2) mmaps the file
+ * 3) create proper input and output streams
+ *
+ * @param filename source filename
+ * @param fec encoder to use
+ */
+int create_coding_files(const char *filename,
+                        nttec::fec::FecCode<uint32_t> *fec) {
+  void *addr;
+  char *tmp_addr;
+  int fd;
+  struct stat sb;
+  size_t length, data_size;
+  off_t offset;
+  char cfilename[1024];
+  std::vector<std::istream *> d_files(fec->n_data, nullptr);
+  std::vector<std::ostream *> c_files(fec->n_outputs, nullptr);
+  std::vector<std::ostream *> c_props_files(fec->n_outputs, nullptr);
+  std::vector<nttec::Properties> c_props(fec->n_outputs);
+
+  fd = open(filename, O_RDONLY);
+  if (fd == -1) {
+    std::cerr << "opening " << filename << " failed\n";
+    return -1;
+  }
+
+  if (fstat(fd, &sb) == -1) {
+    std::cerr << "fstat " << filename << " failed\n";
+    return -1;
+  }
+
+  length = sb.st_size;
+
+  data_size = length / fec->n_data;
+  if (length % fec->n_data != 0) {
+    // FIXME
+    std::cerr << "file size is not multiple of n_data\n";
+    return -1;
+  }
+  std::cerr << "data_size " << data_size << "\n";
+
+  addr = mmap(NULL, length, PROT_READ, MAP_PRIVATE, fd, 0);
+  if (addr == MAP_FAILED) {
+    std::cerr << "mmap failed\n";
+    exit(1);
+  }
+  std::cerr << "length " << length << " addr " << addr << "\n";
+
+  // do it
+  offset = 0;
+  for (u_int32_t i = 0; i < fec->n_data; i++) {
+    if (vflag)
+      std::cerr << "create: opening data " << i << " offset " << offset << "\n";
+    tmp_addr = static_cast<char *>(addr) + offset;
+    d_files[i] = new imemstream(tmp_addr, data_size);
+    offset += data_size;
+  }
+
+  for (u_int32_t i = 0; i < fec->n_outputs; i++) {
+    snprintf(cfilename, sizeof(cfilename), "%s.c%d", filename, i);
+    if (vflag)
+      std::cerr << "create: opening coding for writing " << filename << "\n";
+    c_files[i] = new std::ofstream(cfilename);
+    snprintf(cfilename, sizeof(cfilename), "%s.c%d.props", filename, i);
+    if (vflag)
+      std::cerr << "create: opening coding props for writing " << filename
+                << "\n";
+    c_props_files[i] = new std::ofstream(cfilename);
+  }
+
+  fec->encode_bufs(d_files, c_files, c_props);
+
+  if (munmap(addr, length) != 0) {
+    std::cerr << "munmap " << filename << " failed\n";
+    exit(1);
+  }
+
+  close(fd);
+
+  return 0;
+}
+
+void do_put(const char *filename, int n_data, int n_parities) {
+  nttec::fec::RsFnt<uint32_t> *fec;
+
+  fec = new nttec::fec::RsFnt<uint32_t>(2, n_data, n_parities);
+  if (create_coding_files(filename, fec) != 0) {
+    return;
+  }
+}

src/io_client.h

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2017-2018 the QuadIron authors
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above copyright notice,
+ *    this list of conditions and the following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above copyright notice,
+ *    this list of conditions and the following disclaimer in the documentation
+ *    and/or other materials provided with the distribution.
+ *
+ * 3. Neither the name of the copyright holder nor the names of its contributors
+ *    may be used to endorse or promote products derived from this software
+ *    without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+#ifndef __IO_CLIENT_H__
+#define __IO_CLIENT_H__
+
+#include <nttec/nttec.h>
+
+extern int create_coding_files(const char *filename,
+                               nttec::fec::FecCode<uint32_t> *fec);
+extern void do_put(const char *filename, int n_data, int n_parities);
+
+#endif

src/io_server.cpp

@@ -0,0 +1,60 @@
+
+#include <cstdio>
+#include <cstdlib>
+
+#include <nttec/nttec.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <netinet/ip.h>
+
+#include "io_server.h"
+
+namespace io {
+
+  Server::Server() {
+    this->daemon = nullptr;
+  }
+
+  /**
+   * Get the I/O Server port
+   *
+   * return the port number or 0 on failure
+   */
+  uint16_t Server::get_port()
+  {
+    const union MHD_DaemonInfo *dinfo;
+
+    if (this->daemon == nullptr) {
+      std::cerr << "start daemon first\n";
+      return 0;
+    }
+
+    dinfo = MHD_get_daemon_info(this->daemon, MHD_DAEMON_INFO_LISTEN_FD);
+
+    struct sockaddr_in sin;
+    socklen_t len = sizeof(sin);
+    if (getsockname(dinfo->listen_fd, reinterpret_cast<struct sockaddr *>(&sin), &len) == -1) {
+      std::cerr << "getsockname error\n";
+      return 0;
+    }
+
+    return ntohs(sin.sin_port);
+  }
+
+  int Server::callback(void *, MHD_Connection *, const char *,
+                       const char *, const char *,
+                       const char *, size_t *,
+                       void **) {
+    std::cerr << "callback\n";
+    return 0;
+  }
+
+  void Server::start_daemon() {
+    unsigned int mhd_flags = MHD_NO_FLAG;
+    uint16_t port = 0;
+
+    this->daemon = MHD_start_daemon(mhd_flags, port, NULL, NULL,
+                                    Server::callback, this,
+                                    MHD_OPTION_END);
+  }
+}

src/io_server.h

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2017-2018 the QuadIron authors
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above copyright notice,
+ *    this list of conditions and the following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above copyright notice,
+ *    this list of conditions and the following disclaimer in the documentation
+ *    and/or other materials provided with the distribution.
+ *
+ * 3. Neither the name of the copyright holder nor the names of its contributors
+ *    may be used to endorse or promote products derived from this software
+ *    without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+#ifndef __IO_SERVER_H__
+#define __IO_SERVER_H__
+
+#include <jsonrpccpp/server.h>
+#include <jsonrpccpp/server/connectors/httpserver.h>
+#include <nttec/nttec.h>
+
+namespace io {
+
+  class Server {
+public:
+  Server();
+  ~Server();
+
+  void start_daemon();
+  uint16_t get_port();
+
+private:
+  struct MHD_Daemon *daemon;
+
+  static int callback(void *cls, MHD_Connection *connection, const char *url,
+                      const char *method, const char *version,
+                      const char *upload_data, size_t *upload_data_size,
+                      void **con_cls);
+};
+
+} // namespace io
+
+#endif