<commit_before>/* -*- c-basic-offset: 2 -*- */
/* Copyright(C) 2010-2012 Brazil

  This library is free software; you can redistribute it and/or
  modify it under the terms of the GNU Lesser General Public
  License version 2.1 as published by the Free Software Foundation.

  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  Lesser General Public License for more details.

  You should have received a copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
*/
#include <zmq.h>
#include <stdio.h>
#include <signal.h>
#include <unistd.h>
#include <msgpack.h>
#include <pthread.h>
#include <groonga.h>
#include <inttypes.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <dirent.h>

#include "util.h"

#include <evhttp.h>

#define DEFAULT_RECV_ENDPOINT "tcp://*:1234"
#define DEFAULT_SEND_ENDPOINT "tcp://*:1235"
#define SEND_WAIT 1000 /* 0.001sec */

#define CONST_STR_LEN(x) x, x ? sizeof(x) - 1 : 0

typedef struct {
  const char *db_path;
  const char *send_endpoint;
  pthread_t thd;
  void *zmq_ctx;
} send_thd_data;

static volatile sig_atomic_t loop = 1;

static void
load_to_groonga(grn_ctx *ctx,
                grn_obj *buf,
                const char *query, uint32_t query_len,
                const char *client_id, uint32_t client_id_len,
                const char *learn_target_name, uint32_t learn_target_name_len,
                uint64_t millisec,
                int submit)
{
  GRN_BULK_REWIND(buf);
  GRN_TEXT_PUTS(ctx, buf, "load --table event_");
  GRN_TEXT_PUT(ctx, buf, learn_target_name, learn_target_name_len);
  GRN_TEXT_PUTS(ctx, buf, " --each 'suggest_preparer(_id,type,item,sequence,time,pair_");
  GRN_TEXT_PUT(ctx, buf, learn_target_name, learn_target_name_len);
  GRN_TEXT_PUTS(ctx, buf, ")'");
  grn_ctx_send(ctx, GRN_TEXT_VALUE(buf), GRN_TEXT_LEN(buf), GRN_CTX_MORE);
  grn_ctx_send(ctx, CONST_STR_LEN("["), GRN_CTX_MORE);

  GRN_BULK_REWIND(buf);
  GRN_TEXT_PUTS(ctx, buf, "{\"item\":");
  grn_text_esc(ctx, buf, query, query_len);
  GRN_TEXT_PUTS(ctx, buf, ",\"sequence\":");
  grn_text_esc(ctx, buf, client_id, client_id_len);
  GRN_TEXT_PUTS(ctx, buf, ",\"time\":");
  grn_text_ftoa(ctx, buf, (double)millisec / 1000);
  if (submit) {
    GRN_TEXT_PUTS(ctx, buf, ",\"type\":\"submit\"}");
  } else {
    GRN_TEXT_PUTS(ctx, buf, "}");
  }
  /* printf("%.*s\n", GRN_TEXT_LEN(buf), GRN_TEXT_VALUE(buf)); */
  grn_ctx_send(ctx, GRN_TEXT_VALUE(buf), GRN_TEXT_LEN(buf), GRN_CTX_MORE);

  grn_ctx_send(ctx, CONST_STR_LEN("]"), 0);

  {
    char *res;
    int flags;
    unsigned int res_len;
    grn_ctx_recv(ctx, &res, &res_len, &flags);
  }
}

void
load_to_multi_targets(grn_ctx *ctx,
                      grn_obj *buf,
                      const char *query, uint32_t query_len,
                      const char *client_id, uint32_t client_id_len,
                      const char *learn_target_names,
                      uint32_t learn_target_names_len,
                      uint64_t millisec,
                      int submit)
{
  if (millisec && query && client_id && learn_target_names) {
    unsigned int tn_len;
    const char *tn, *tnp, *tne;
    tn = tnp = learn_target_names;
    tne = learn_target_names + learn_target_names_len;
    while (tnp <= tne) {
      if (tnp == tne || *tnp == '|') {
        tn_len = tnp - tn;

        /*
        printf("sec: %" PRIu64 " query %.*s client_id: %.*s target: %.*s\n",
          millisec,
          query_len, query,
          client_id_len, client_id,
          tn_len, tn);
        */
        load_to_groonga(ctx, buf, query, query_len, client_id, client_id_len,
                        tn, tn_len, millisec, submit);

        tn = ++tnp;
      } else {
        tnp++;
      }
    }
  }
}

#define PACK_KEY_FROM_ID(id) do { \
  int _k_len; \
  char _k_buf[GRN_TABLE_MAX_KEY_SIZE]; \
  _k_len = grn_table_get_key(ctx, ref_table, (id), _k_buf, GRN_TABLE_MAX_KEY_SIZE); \
  msgpack_pack_raw(&pk, _k_len); \
  msgpack_pack_raw_body(&pk, _k_buf, _k_len); \
} while (0)

#define PACK_MAP_ITEM(col_name) do { \
  grn_obj _v; \
  msgpack_pack_raw(&pk, sizeof(#col_name) - 1); \
  msgpack_pack_raw_body(&pk, CONST_STR_LEN(#col_name)); \
  switch (col_##col_name->header.type) { \
  case GRN_COLUMN_FIX_SIZE: \
    GRN_VALUE_FIX_SIZE_INIT(&_v, 0, grn_obj_get_range(ctx, col_##col_name)); \
    break; \
  case GRN_COLUMN_VAR_SIZE: \
    if ((col_##col_name->header.flags & GRN_OBJ_COLUMN_TYPE_MASK) == GRN_OBJ_COLUMN_VECTOR) { \
      GRN_VALUE_FIX_SIZE_INIT(&_v, GRN_OBJ_VECTOR, grn_obj_get_range(ctx, col_##col_name)); \
    } else { \
      GRN_VALUE_VAR_SIZE_INIT(&_v, 0, grn_obj_get_range(ctx, col_##col_name)); \
    } \
    break; \
  } \
  grn_obj_get_value(ctx, col_##col_name, rec_id, &_v); \
 \
  switch (_v.header.type) { \
  case GRN_BULK: \
    switch (_v.header.domain) { \
    case GRN_DB_SHORT_TEXT: \
      msgpack_pack_raw(&pk, GRN_TEXT_LEN(&_v)); \
      msgpack_pack_raw_body(&pk, GRN_TEXT_VALUE(&_v), GRN_TEXT_LEN(&_v)); \
      break; \
    case GRN_DB_INT32: \
      msgpack_pack_int32(&pk, GRN_INT32_VALUE(&_v)); \
      break; \
    case GRN_DB_UINT32: \
      msgpack_pack_uint32(&pk, GRN_UINT32_VALUE(&_v)); \
      break; \
    case GRN_DB_TIME: \
      msgpack_pack_double(&pk, (double)GRN_TIME_VALUE(&_v) / GRN_TIME_USEC_PER_SEC); \
      break; \
    default: /* ref. to ShortText key */ \
      PACK_KEY_FROM_ID(GRN_RECORD_VALUE(&_v)); \
    } \
    break; \
  case GRN_UVECTOR: /* ref.s to ShortText key */ \
    { \
      grn_id *_idv = (grn_id *)GRN_BULK_HEAD(&_v), *_idve = (grn_id *)GRN_BULK_CURR(&_v); \
      msgpack_pack_array(&pk, _idve - _idv); \
      for (; _idv < _idve; _idv++) { \
        PACK_KEY_FROM_ID(*_idv); \
      } \
    } \
    break; \
  default: \
    print_error("invalid groonga object type(%d) for msgpack.", _v.header.type); \
    msgpack_pack_nil(&pk); \
    break; \
  } \
  grn_obj_close(ctx, &_v); \
} while (0)

static int
zmq_send_to_httpd(void *zmq_send_sock, void *data, size_t size)
{
  zmq_msg_t msg;
  if (!zmq_msg_init_size(&msg, size)) {
    memcpy((void *)zmq_msg_data(&msg), data, size);
    if (zmq_send(zmq_send_sock, &msg, 0)) {
      print_error("zmq_send() error");
      return -1;
    }
    zmq_msg_close(&msg);
  } else {
    print_error("zmq_msg_init_size() error");
  }
  return 0;
}

static void
send_handler(void *zmq_send_sock, grn_ctx *ctx)
{
  grn_table_cursor *cur;
  if ((cur = grn_table_cursor_open(ctx, grn_ctx_db(ctx), NULL, 0, NULL, 0,
       0, -1, 0))) {
    grn_id table_id;
    while (loop && (table_id = grn_table_cursor_next(ctx, cur)) != GRN_ID_NIL) {
      grn_obj *table;
      if ((table = grn_ctx_at(ctx, table_id))) {
        int name_len;
        char name_buf[GRN_TABLE_MAX_KEY_SIZE];

        name_len = grn_obj_name(ctx, table, name_buf,
                                GRN_TABLE_MAX_KEY_SIZE);<commit_msg>suggest: support zmq_msg_send() in zmq >= 3<commit_after>/* -*- c-basic-offset: 2 -*- */
/* Copyright(C) 2010-2013 Brazil

  This library is free software; you can redistribute it and/or
  modify it under the terms of the GNU Lesser General Public
  License version 2.1 as published by the Free Software Foundation.

  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  Lesser General Public License for more details.

  You should have received a copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
*/
#include <zmq.h>
#include <stdio.h>
#include <signal.h>
#include <unistd.h>
#include <msgpack.h>
#include <pthread.h>
#include <groonga.h>
#include <inttypes.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <dirent.h>

#include "util.h"

#include <evhttp.h>

#if ZMQ_VERSION_MAJOR == 2
#  define zmq_msg_send(message, socket, flags) \
  zmq_send((socket), (message), (flags))
#endif

#define DEFAULT_RECV_ENDPOINT "tcp://*:1234"
#define DEFAULT_SEND_ENDPOINT "tcp://*:1235"
#define SEND_WAIT 1000 /* 0.001sec */

#define CONST_STR_LEN(x) x, x ? sizeof(x) - 1 : 0

typedef struct {
  const char *db_path;
  const char *send_endpoint;
  pthread_t thd;
  void *zmq_ctx;
} send_thd_data;

static volatile sig_atomic_t loop = 1;

static void
load_to_groonga(grn_ctx *ctx,
                grn_obj *buf,
                const char *query, uint32_t query_len,
                const char *client_id, uint32_t client_id_len,
                const char *learn_target_name, uint32_t learn_target_name_len,
                uint64_t millisec,
                int submit)
{
  GRN_BULK_REWIND(buf);
  GRN_TEXT_PUTS(ctx, buf, "load --table event_");
  GRN_TEXT_PUT(ctx, buf, learn_target_name, learn_target_name_len);
  GRN_TEXT_PUTS(ctx, buf, " --each 'suggest_preparer(_id,type,item,sequence,time,pair_");
  GRN_TEXT_PUT(ctx, buf, learn_target_name, learn_target_name_len);
  GRN_TEXT_PUTS(ctx, buf, ")'");
  grn_ctx_send(ctx, GRN_TEXT_VALUE(buf), GRN_TEXT_LEN(buf), GRN_CTX_MORE);
  grn_ctx_send(ctx, CONST_STR_LEN("["), GRN_CTX_MORE);

  GRN_BULK_REWIND(buf);
  GRN_TEXT_PUTS(ctx, buf, "{\"item\":");
  grn_text_esc(ctx, buf, query, query_len);
  GRN_TEXT_PUTS(ctx, buf, ",\"sequence\":");
  grn_text_esc(ctx, buf, client_id, client_id_len);
  GRN_TEXT_PUTS(ctx, buf, ",\"time\":");
  grn_text_ftoa(ctx, buf, (double)millisec / 1000);
  if (submit) {
    GRN_TEXT_PUTS(ctx, buf, ",\"type\":\"submit\"}");
  } else {
    GRN_TEXT_PUTS(ctx, buf, "}");
  }
  /* printf("%.*s\n", GRN_TEXT_LEN(buf), GRN_TEXT_VALUE(buf)); */
  grn_ctx_send(ctx, GRN_TEXT_VALUE(buf), GRN_TEXT_LEN(buf), GRN_CTX_MORE);

  grn_ctx_send(ctx, CONST_STR_LEN("]"), 0);

  {
    char *res;
    int flags;
    unsigned int res_len;
    grn_ctx_recv(ctx, &res, &res_len, &flags);
  }
}

void
load_to_multi_targets(grn_ctx *ctx,
                      grn_obj *buf,
                      const char *query, uint32_t query_len,
                      const char *client_id, uint32_t client_id_len,
                      const char *learn_target_names,
                      uint32_t learn_target_names_len,
                      uint64_t millisec,
                      int submit)
{
  if (millisec && query && client_id && learn_target_names) {
    unsigned int tn_len;
    const char *tn, *tnp, *tne;
    tn = tnp = learn_target_names;
    tne = learn_target_names + learn_target_names_len;
    while (tnp <= tne) {
      if (tnp == tne || *tnp == '|') {
        tn_len = tnp - tn;

        /*
        printf("sec: %" PRIu64 " query %.*s client_id: %.*s target: %.*s\n",
          millisec,
          query_len, query,
          client_id_len, client_id,
          tn_len, tn);
        */
        load_to_groonga(ctx, buf, query, query_len, client_id, client_id_len,
                        tn, tn_len, millisec, submit);

        tn = ++tnp;
      } else {
        tnp++;
      }
    }
  }
}

#define PACK_KEY_FROM_ID(id) do { \
  int _k_len; \
  char _k_buf[GRN_TABLE_MAX_KEY_SIZE]; \
  _k_len = grn_table_get_key(ctx, ref_table, (id), _k_buf, GRN_TABLE_MAX_KEY_SIZE); \
  msgpack_pack_raw(&pk, _k_len); \
  msgpack_pack_raw_body(&pk, _k_buf, _k_len); \
} while (0)

#define PACK_MAP_ITEM(col_name) do { \
  grn_obj _v; \
  msgpack_pack_raw(&pk, sizeof(#col_name) - 1); \
  msgpack_pack_raw_body(&pk, CONST_STR_LEN(#col_name)); \
  switch (col_##col_name->header.type) { \
  case GRN_COLUMN_FIX_SIZE: \
    GRN_VALUE_FIX_SIZE_INIT(&_v, 0, grn_obj_get_range(ctx, col_##col_name)); \
    break; \
  case GRN_COLUMN_VAR_SIZE: \
    if ((col_##col_name->header.flags & GRN_OBJ_COLUMN_TYPE_MASK) == GRN_OBJ_COLUMN_VECTOR) { \
      GRN_VALUE_FIX_SIZE_INIT(&_v, GRN_OBJ_VECTOR, grn_obj_get_range(ctx, col_##col_name)); \
    } else { \
      GRN_VALUE_VAR_SIZE_INIT(&_v, 0, grn_obj_get_range(ctx, col_##col_name)); \
    } \
    break; \
  } \
  grn_obj_get_value(ctx, col_##col_name, rec_id, &_v); \
 \
  switch (_v.header.type) { \
  case GRN_BULK: \
    switch (_v.header.domain) { \
    case GRN_DB_SHORT_TEXT: \
      msgpack_pack_raw(&pk, GRN_TEXT_LEN(&_v)); \
      msgpack_pack_raw_body(&pk, GRN_TEXT_VALUE(&_v), GRN_TEXT_LEN(&_v)); \
      break; \
    case GRN_DB_INT32: \
      msgpack_pack_int32(&pk, GRN_INT32_VALUE(&_v)); \
      break; \
    case GRN_DB_UINT32: \
      msgpack_pack_uint32(&pk, GRN_UINT32_VALUE(&_v)); \
      break; \
    case GRN_DB_TIME: \
      msgpack_pack_double(&pk, (double)GRN_TIME_VALUE(&_v) / GRN_TIME_USEC_PER_SEC); \
      break; \
    default: /* ref. to ShortText key */ \
      PACK_KEY_FROM_ID(GRN_RECORD_VALUE(&_v)); \
    } \
    break; \
  case GRN_UVECTOR: /* ref.s to ShortText key */ \
    { \
      grn_id *_idv = (grn_id *)GRN_BULK_HEAD(&_v), *_idve = (grn_id *)GRN_BULK_CURR(&_v); \
      msgpack_pack_array(&pk, _idve - _idv); \
      for (; _idv < _idve; _idv++) { \
        PACK_KEY_FROM_ID(*_idv); \
      } \
    } \
    break; \
  default: \
    print_error("invalid groonga object type(%d) for msgpack.", _v.header.type); \
    msgpack_pack_nil(&pk); \
    break; \
  } \
  grn_obj_close(ctx, &_v); \
} while (0)

static int
zmq_send_to_httpd(void *zmq_send_sock, void *data, size_t size)
{
  zmq_msg_t msg;
  if (!zmq_msg_init_size(&msg, size)) {
    memcpy((void *)zmq_msg_data(&msg), data, size);
    if (zmq_msg_send(&msg, zmq_send_sock, 0)) {
      print_error("zmq_send() error");
      return -1;
    }
    zmq_msg_close(&msg);
  } else {
    print_error("zmq_msg_init_size() error");
  }
  return 0;
}

static void
send_handler(void *zmq_send_sock, grn_ctx *ctx)
{
  grn_table_cursor *cur;
  if ((cur = grn_table_cursor_open(ctx, grn_ctx_db(ctx), NULL, 0, NULL, 0,
       0, -1, 0))) {
    grn_id table_id;
    while (loop && (table_id = grn_table_cursor_next(ctx, cur)) != GRN_ID_NIL) {
      grn_obj *table;
      if ((table = grn_ctx_at(ctx, table_id))) {
        int name_len;
        char name_buf[GRN_TABLE_MAX_KEY_SIZE];

        name_len = grn_obj_name(ctx, table, name_buf,
                                GRN_TABLE_MAX_KEY_SIZE);
{'fraction_non_alphanumeric': 0.13409961685823754, 'fraction_numerical': 0.012539184952978056, 'mean_word_length': 3.03939223410242, 'pattern_counts': {'":': 8, '<': 31, '<?xml version=': 0, '>': 32, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5092484', 'n_tokens_mistral': 5896, 'n_tokens_neox': 5685, 'n_words': 1139}