text: /* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
  
  /*  Chunk I/O
   *  =========
   *  Copyright 2018 Eduardo Silva <eduardo@monkey.io>
   *
   *  Licensed under the Apache License, Version 2.0 (the "License");
   *  you may not use this file except in compliance with the License.
   *  You may obtain a copy of the License at
   *
   *      http://www.apache.org/licenses/LICENSE-2.0
   *
   *  Unless required by applicable law or agreed to in writing, software
   *  distributed under the License is distributed on an "AS IS" BASIS,
   *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   *  See the License for the specific language governing permissions and
   *  limitations under the License.
   */
  
  #include <stdio.h>
  #include <stdlib.h>
  #include <unistd.h>
  #include <getopt.h>
  #include <string.h>
  #include <sys/types.h>
  #include <sys/mman.h>
  #include <sys/time.h>
  #include <pwd.h>
  #include <limits.h>
  #include <signal.h>
  #include <time.h>
  #ifdef __MACH__
  #  include <mach/clock.h>
  #  include <mach/mach.h>
  #endif
  
  #define ANSI_RESET "\033[0m"
  #define ANSI_BOLD  "\033[1m"
  
  #define ANSI_CYAN          "\033[96m"
  #define ANSI_BOLD_CYAN     ANSI_BOLD ANSI_CYAN
  #define ANSI_MAGENTA       "\033[95m"
  #define ANSI_BOLD_MAGENTA  ANSI_BOLD ANSI_MAGENTA
  #define ANSI_RED           "\033[91m"
  #define ANSI_BOLD_RED      ANSI_BOLD ANSI_RED
  #define ANSI_YELLOW        "\033[93m"
  
  #define ANSI_BOLD_YELLOW   ANSI_BOLD ANSI_YELLOW
  #define ANSI_BLUE          "\033[94m"
  #define ANSI_BOLD_BLUE     ANSI_BOLD ANSI_BLUE
  #define ANSI_GREEN         "\033[92m"
  #define ANSI_BOLD_GREEN    ANSI_BOLD ANSI_GREEN
  #define ANSI_WHITE         "\033[97m"
  #define ANSI_BOLD_WHITE    ANSI_BOLD ANSI_WHITE
  
  #define CIO_ROOT_PATH  ".cio"
  #define cio_print_signal(X) case X:                       \
      write (STDERR_FILENO, #X ")\n" , sizeof(#X ")\n")-1); \
      break;
  
  #define CIO_PERF_PATH      "/tmp/cio-perf/"
  
  #define ONESEC_IN_NSEC     1000000000
  
  #include <chunkio/chunkio.h>
  #include <chunkio/cio_log.h>
  #include <chunkio/cio_stream.h>
  #include <chunkio/cio_chunk.h>
  #include <chunkio/cio_meta.h>
  #include <chunkio/cio_scan.h>
  #include <chunkio/cio_utils.h>
  
  static void cio_help(int rc)
  {
      printf("Usage: cio [-r PATH] OPTIONS\n\n");
      printf("Available Options\n");
      printf("  -r, --root[=PATH]\tset root path\n");
      printf("  -i, --stdin\t\tdump stdin data to stream/file\n");
      printf("  -s, --stream=STREAM\tset stream name\n");
      printf("  -m, --metadata=INFO\tset metadata\n");
      printf("  -M, --memory\t\trun in-memory mode\n");
      printf("  -l, --list\t\tlist environment content\n");
      printf("  -F, --full-sync\tforce data flush to disk\n");
      printf("  -k, --checksum\tenable CRC32 checksum\n");
      printf("  -p, --perf=FILE\trun performance test\n");
      printf("  -w, --perf-writes=N\tset number of writes for performance mode "
             "(default: 5)\n");
      printf("  -f, --perf-files=N\tset number of files to create on "
             "performance mode (default: 1000)\n");
      printf("  -h, --help\t\tprint this help\n");
      exit(rc);
  }
  
  static void cio_signal_handler(int signal)
  {
      char s[] = "[cio] caught signal (";
  
      /* write signal number */
      write(STDERR_FILENO, s, sizeof(s) - 1);
      switch (signal) {
          cio_print_signal(SIGINT);
          cio_print_signal(SIGQUIT);
          cio_print_signal(SIGHUP);
          cio_print_signal(SIGTERM);
          cio_print_signal(SIGSEGV);
      };
  
      /* Signal handlers */
      switch (signal) {
      case SIGINT:
      case SIGQUIT:
      case SIGHUP:
      case SIGTERM:
          _exit(EXIT_SUCCESS);
      case SIGSEGV:
          abort();
      default:
          break;
      }
  }
  
  void cio_bytes_to_human_readable_size(size_t bytes,
                                        char *out_buf, size_t size)
  {
      unsigned long i;
      unsigned long u = 1024;
      static const char *__units[] = {
          "b", "K", "M", "G",
          "T", "P", "E", "Z", "Y", NULL
      };
  
      for (i = 0; __units[i] != NULL; i++) {
          if ((bytes / u) == 0) {
              break;
          }
          u *= 1024;
      }
      if (!i) {
          snprintf(out_buf, size, "%lu%s", (long unsigned int) bytes, __units[0]);
      }
      else {
          float fsize = (float) ((double) bytes / (u / 1024));
          snprintf(out_buf, size, "%.1f%s", fsize, __units[i]);
      }
  }
  
  static void cio_signal_init()
  {
      signal(SIGINT,  &cio_signal_handler);
      signal(SIGQUIT, &cio_signal_handler);
      signal(SIGHUP,  &cio_signal_handler);
      signal(SIGTERM, &cio_signal_handler);
      signal(SIGSEGV, &cio_signal_handler);
  }
  
  static int log_cb(struct cio_ctx *ctx, const char *file, int line,
                    char *str)
  {
      char *dtitle = "chunkio";
      char *dcolor = ANSI_BLUE;
  
      /* messages from this own client are in yellow */
      if (*file == 't') {
          dtitle = "  cli  ";
          dcolor = ANSI_YELLOW;
      }
  
      if (ctx->log_level > CIO_INFO) {
          printf("%s[%s]%s %-60s => %s%s:%i%s\n",
                 dcolor, dtitle, ANSI_RESET, str,
                 dcolor, file, line, ANSI_RESET);
      }
      else {
          printf("%s[%s]%s %s\n", dcolor, dtitle, ANSI_RESET, str);
      }
  
      return 0;
  }
  
  static int cio_default_root_path(char *path, int size)
  {
      int len;
      struct passwd *pw;
  
      pw = getpwuid(getuid());
      if (!pw) {
          perror("getpwuid");
          return -1;
      }
  
      /* ~/.cio */
      len = snprintf(path, size, "%s/%s",
                     pw->pw_dir, CIO_ROOT_PATH);
      if (len == -1) {
          perror("snprintf");
          return -1;
      }
  
      return 0;
  }
  
  static void cio_timespec_get(struct timespec *t)
  {
  #ifdef __MACH__ // macOS does not have timespec_get, use clock_get_time
      clock_serv_t cclock;
      mach_timespec_t mts;
      host_get_clock_service(mach_host_self(), CALENDAR_CLOCK, &cclock);
      clock_get_time(cclock, &mts);
      mach_port_deallocate(mach_task_self(), cclock);
      t->tv_sec = mts.tv_sec;
      t->tv_nsec = mts.tv_nsec;
  #else
      timespec_get(t, TIME_UTC);
  #endif
  }
  
  /* command/list: iterate root path and list content */
  static int cb_cmd_list(struct cio_ctx *ctx)
  {
      cio_scan_dump(ctx);
      return 0;
  }
  
  /* command/stdin: read data from STDIN and dump it into stream/file */
  static int cb_cmd_stdin(struct cio_ctx *ctx, const char *stream,
                          int opt_buffer,
                          const char *fname, const char *metadata)
  {
      int fd;
      int ret;
      int meta_len;
      size_t total = 0;
      ssize_t bytes;
      char buf[1024*8];
      struct cio_stream *st;
      struct cio_chunk *ch;
  
      /* Prepare stream and file contexts */
      st = cio_stream_create(ctx, stream, opt_buffer);
      if (!st) {
          cio_log_error(ctx, "cannot create stream\n");
          return -1;
      }
  
      /* Open a file with a hint of 32KB */
      ch = cio_chunk_open(ctx, st, fname, CIO_OPEN, 1024*32);
      if (!ch) {
          cio_log_error(ctx, "cannot create file");
          return -1;
      }
  
      if (metadata) {
          meta_len = strlen(metadata);
          cio_meta_write(ch, (char *) metadata, meta_len);
      }
  
      /* Catch up stdin */
      fd = dup(STDIN_FILENO);
      if (fd == -1) {
          perror("dup");
          cio_log_error(ctx, "cannot open standard input");
          return -1;
      }
  
      do {
          bytes = read(fd, buf, sizeof(buf) - 1);
          if (bytes == 0) {
              break;
          }
          else if (bytes == -1) {
              perror("read");
          }
          else {
              ret = cio_chunk_write(ch, buf, bytes);
              if (ret == -1) {
                  cio_log_error(ctx, "error writing to file");
                  close(fd);
                  return -1;
              }
              total += bytes;
          }
      } while (bytes > 0);
  
      /* close stdin dup(2) */
      close(fd);
  
      /* synchronize changes to disk and close */
      cio_chunk_sync(ch);
      cio_chunk_close(ch, CIO_FALSE);
  
      /* print some status */
      cio_bytes_to_human_readable_size(total, buf, sizeof(buf) - 1);
      cio_log_info(ctx, "stdin total bytes => %lu (%s)", total, buf);
  
      return 0;
  }
  
  static double time_to_double(struct timespec *t)
  {
      return (double)(t->tv_sec) + ((double)t->tv_nsec/(double) ONESEC_IN_NSEC);
  }
  
  static int time_diff(struct timespec *tm0, struct timespec *tm1,
                       struct timespec *out)
  {
  
      if (tm1->tv_sec >= tm0->tv_sec) {
          out->tv_sec = tm1->tv_sec - tm0->tv_sec;
          if (tm1->tv_nsec >= tm0->tv_nsec) {
              out->tv_nsec = tm1->tv_nsec - tm0->tv_nsec;
          }
          else if(tm0->tv_sec == 0){
              /* underflow */
              return -1;
          }
          else{
              out->tv_nsec = ONESEC_IN_NSEC \
                  + tm1->tv_nsec - tm0->tv_nsec;
              out->tv_sec--;
          }
      }
      else {
          /* underflow */
          return -1;
      }
  
      return 0;
  }
  
  static void cb_cmd_perf(struct cio_ctx *ctx, int opt_buffer, char *pfile,
                          char *metadata, int writes, int files)
  {
      int i;
      int j;
      int meta_len = 0;
      int ret;
      uint64_t bytes = 0;
      double rate;
      char *in_data;
      size_t in_size;
      char tmp[255];
      struct cio_stream *stream;
      struct cio_chunk **carr;
      struct timespec t1;
      struct timespec t2;
      struct timespec t_final;
  
      /* Create pref stream */
      stream = cio_stream_create(ctx, "test-perf", opt_buffer);
  
      /*
       * Load sample data file and with the same content through multiple write
       * operations generating other files.
       */
      ret = cio_utils_read_file(pfile, &in_data, &in_size);
      if (ret == -1) {
          cio_destroy(ctx);
          exit(EXIT_FAILURE);
      }
  
      /* Allocate files array */
      carr = calloc(1, sizeof(struct cio_chunk) * files);
      if (!carr) {
          perror("calloc");
          exit(EXIT_FAILURE);
      }
  
      if (metadata) {
          meta_len = strlen(metadata);
      }
  
      /* Perf-write test */
      cio_timespec_get(&t1);
      for (i = 0; i < files; i++) {
          snprintf(tmp, sizeof(tmp), "perf-test-%04i.txt", i);
          carr[i] = cio_chunk_open(ctx, stream, tmp, CIO_OPEN, in_size);
          if (carr[i] == NULL) {
              continue;
          }
  
          if (meta_len > 0) {
              cio_meta_write(carr[i], metadata, meta_len);
              bytes += meta_len;
          }
  
          for (j = 0; j < writes; j++) {
              ret = cio_chunk_write(carr[i], in_data, in_size);
              if (ret == -1) {
                  exit(1);
              }
              bytes += in_size;
          }
          cio_chunk_sync(carr[i]);
          cio_chunk_close(carr[i], CIO_FALSE);
      }
      cio_timespec_get(&t2);
  
      /* Check timing */
      time_diff(&t1, &t2, &t_final);
  
      /*
       * Write out the report
       * ====================
       */
      cio_bytes_to_human_readable_size(bytes, tmp, sizeof(tmp) - 1);
      printf("=== perf write === \n");
      printf("-  crc32 checksum : %s\n",
             ctx->flags & CIO_CHECKSUM ? "enabled" : "disabled");
      printf("-  fs sync mode   : %s\n",
             ctx->flags & CIO_FULL_SYNC ? "full" : "normal");
  
      cio_bytes_to_human_readable_size(in_size, tmp, sizeof(tmp) - 1);
      printf("-  file size      : %s (%lu bytes)\n", tmp, in_size);
      printf("-  total files    : %i\n", files);
      printf("-  file writes    : %i\n", writes);
  
      cio_bytes_to_human_readable_size(bytes, tmp, sizeof(tmp) - 1);
      printf("-  bytes written  : %s (%lu bytes)\n" , tmp, bytes);
      printf("-  elapsed time   : %.2f seconds\n", time_to_double(&t_final));
  
      rate = (double) (bytes / time_to_double(&t_final));
      cio_bytes_to_human_readable_size(rate, tmp, sizeof(tmp) - 1);
      printf("-  rate           : %s per second (%.2f bytes)\n", tmp, rate);
  
      /* Release file data and destroy context */
      free(carr);
      munmap(in_data, in_size);
  }
  
  int main(int argc, char **argv)
  {
      int ret;
      int opt;
      int opt_silent = CIO_FALSE;
      int opt_pwrites = 5;
      int opt_pfiles = 1000;
      int opt_buffer = CIO_STORE_FS;
      int cmd_stdin = CIO_FALSE;
      int cmd_list = CIO_FALSE;
      int cmd_perf = CIO_FALSE;
      int verbose = CIO_WARN;
      int flags = 0;
      char *perf_file = NULL;
      char *fname = NULL;
      char *stream = NULL;
      char *metadata = NULL;
      char *root_path = NULL;
      char tmp[PATH_MAX];
      struct cio_ctx *ctx;
  
      static const struct option long_opts[] = {
          {"full-sync"  , no_argument      , NULL, 'F'},
          {"checksum"   , no_argument      , NULL, 'k'},
          {"list"       , no_argument      , NULL, 'l'},
          {"root"       , required_argument, NULL, 'r'},
          {"silent"     , no_argument      , NULL, 'S'},
          {"stdin"      , no_argument      , NULL, 'i'},
          {"stream"     , required_argument, NULL, 's'},
          {"metadata"   , required_argument, NULL, 'm'},
          {"memory"     , no_argument      , NULL, 'M'},
          {"filename"   , required_argument, NULL, 'f'},
          {"perf"       , required_argument, NULL, 'p'},
          {"perf-writes", required_argument, NULL, 'w'},
          {"perf-files" , required_argument, NULL, 'e'},
          {"verbose"    , no_argument      , NULL, 'v'},
          {"help"       , no_argument      , NULL, 'h'},
      };
  
      /* Initialize signals */
      cio_signal_init();
  
      while ((opt = getopt_long(argc, argv, "Fklr:p:w:e:Sis:m:Mf:vh",
                                long_opts, NULL)) != -1) {
          switch (opt) {
          case 'F':
              flags |= CIO_FULL_SYNC;
              break;
          case 'k':
              flags |= CIO_CHECKSUM;
              break;
          case 'l':
              cmd_list = CIO_TRUE;
              break;
          case 'i':
              cmd_stdin = CIO_TRUE;
              break;
          case 'r':
              root_path = strdup(optarg);
              break;
          case 'p':
              cio_utils_recursive_delete(CIO_PERF_PATH);
              perf_file = strdup(optarg);
              root_path = strdup(CIO_PERF_PATH);
              cmd_perf = CIO_TRUE;
              break;
          case 'w':
              opt_pwrites = atoi(optarg);
              break;
          case 'e':
              opt_pfiles = atoi(optarg);
              break;
          case 'S':
              opt_silent = CIO_TRUE;
              break;
          case 's':
              stream = strdup(optarg);
              break;
          case 'm':
              metadata = strdup(optarg);
              break;
          case 'M':
              opt_buffer = CIO_STORE_MEM;
              break;
          case 'f':
              fname = strdup(optarg);
              break;
          case 'v':
              verbose++;
              break;
          case 'h':
              cio_help(EXIT_SUCCESS);
              break;
          default:
              cio_help(EXIT_FAILURE);
          }
      }
  
      /* Check root path, if not set, defaults to ~/.cio */
      if (!root_path) {
          ret = cio_default_root_path(tmp, sizeof(tmp) - 1);
          if (ret == -1) {
              fprintf(stderr,
                      "[chunkio cli] cannot set default root path\n");
              cio_help(EXIT_FAILURE);
          }
          root_path = strdup(tmp);
      }
  
      if (opt_silent == CIO_TRUE) {
          verbose = 0;
      }
  
      /* Create CIO instance */
      ctx = cio_create(root_path, log_cb, verbose, flags);
      free(root_path);
  
      if (!ctx) {
          exit(EXIT_FAILURE);
      }
  
      cio_log_info(ctx, "root_path => %s", ctx->root_path);
  
      /*
       * Process commands and options
       */
      if (cmd_list == CIO_TRUE) {
          ret = cb_cmd_list(ctx);
      }
      else if (cmd_stdin == CIO_TRUE) {
          /* we need the stream and file names */
          if (!stream) {
              fprintf(stderr, "[chunkio cli] missing stream name\n");
              cio_help(EXIT_FAILURE);
          }
          if (!fname) {
              fprintf(stderr, "[chunkio cli] missing file name\n");
              free(stream);
              cio_help(EXIT_FAILURE);
          }
  
          ret = cb_cmd_stdin(ctx, stream, opt_buffer, fname, metadata);
      }
      else if (cmd_perf == CIO_TRUE) {
          cb_cmd_perf(ctx, opt_buffer, perf_file, metadata, opt_pwrites,
                      opt_pfiles);
          free(perf_file);
      }
      else {
          cio_help(EXIT_FAILURE);
      }
  
      free(stream);
      free(fname);
      free(metadata);
      cio_destroy(ctx);
  
      return ret;
  }
  