library/srw.c

   1 /*
   2  * 2011+ Copyright (c) Evgeniy Polyakov <zbr@ioremap.net>
   3  * All rights reserved.
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License as published by
   7  * the Free Software Foundation; either version 2 of the License, or
   8  * (at your option) any later version.
   9  *
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License for more details.
  14  */
  15
  16 #include "config.h"
  17
  18 #include <sys/types.h>
  19 #include <sys/stat.h>
  20 #include <sys/socket.h>
  21 #include <sys/mman.h>
  22 #include <sys/wait.h>
  23
  24 #include <ctype.h>
  25 #include <fcntl.h>
  26 #include <limits.h>
  27 #include <stdio.h>
  28 #include <stdlib.h>
  29 #include <unistd.h>
  30
  31 #include "elliptics.h"
  32
  33 #include <elliptics/packet.h>
  34 #include <elliptics/interface.h>
  35
  36 #include <elliptics/srw/srwc.h>
  37
  38 int dnet_srw_init(struct dnet_node *n, struct dnet_config *cfg)
  39 {
  40         int err = 0;
  41
  42         if (!cfg->srw.config)
  43                 cfg->srw.config = cfg->addr;
  44
  45         dnet_log(n, DNET_LOG_INFO, "srw: binary: '%s', log: '%s', pipe: '%s', init: '%s', config: '%s', threads: %d\n",
  46                         cfg->srw.binary, cfg->srw.log, cfg->srw.pipe, cfg->srw.init, cfg->srw.config, cfg->srw.num);
  47
  48         if (!cfg->srw.init || cfg->srw.num <= 0 || !cfg->srw.binary || !cfg->srw.pipe) {
  49                 err = 0;
  50                 dnet_log(n, DNET_LOG_INFO, "srw: do not initialize - insufficient parameters in config\n");
  51                 goto err_out_exit;
  52         }
  53
  54         cfg->srw.priv = NULL;
  55
  56         n->srw = srwc_init(&cfg->srw);
  57         if (!n->srw) {
  58                 err = -EINVAL;
  59                 dnet_log(n, DNET_LOG_ERROR, "srw: failed to initialize external python workers\n");
  60                 goto err_out_exit;
  61         }
  62
  63 err_out_exit:
  64         return err;
  65 }
  66
  67 void dnet_srw_cleanup(struct dnet_node *n)
  68 {
  69         if (n->srw)
  70                 srwc_cleanup(n->srw);
  71 }
  72
  73 int dnet_cmd_exec_raw(struct dnet_net_state *st, struct dnet_cmd *cmd, struct dnet_attr *attr,
  74                 struct sph *header, const void *data)
  75 {
  76         struct dnet_node *n = st->n;
  77         int err;
  78         struct srwc_ctl ctl;
  79
  80         memset(&ctl, 0, sizeof(struct srwc_ctl));
  81
  82         ctl.header = *header;
  83
  84         err = srwc_process(n->srw, &ctl, data);
  85         if (err < 0) {
  86                 dnet_log(n, DNET_LOG_ERROR, "%s: python processing failed: %s %d\n", dnet_dump_id(&cmd->id), strerror(-err), err);
  87                 goto err_out_exit;
  88         }
  89
  90         dnet_log(n, DNET_LOG_DSA, "%s: reply %llu bytes: '%.*s'\n",
  91                         dnet_dump_id(&cmd->id), (unsigned long long)ctl.res_size, (int)ctl.res_size, ctl.result);
  92
  93         if (ctl.res_size) {
  94                 err = dnet_send_reply(st, cmd, attr, ctl.result, ctl.res_size, 0);
  95                 free(ctl.result);
  96         }
  97
  98 err_out_exit:
  99         return err;
 100 }
 101
 102 int dnet_srw_update(struct dnet_node *n, int pid)
 103 {
 104         return srwc_drop(n->srw, pid);
 105 }