1/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
2
3/**
4 * @copyright 2013 Couchbase, Inc.
5 *
6 * @author Sarath Lakshman  <sarath@couchbase.com>
7 *
8 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
9 * use this file except in compliance with the License. You may obtain a copy of
10 * the License at
11 *
12 *  http://www.apache.org/licenses/LICENSE-2.0
13 *
14 * Unless required by applicable law or agreed to in writing, software
15 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
16 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
17 * License for the specific language governing permissions and limitations under
18 * the License.
19 **/
20
21#include "config.h"
22
23#include <platform/cb_malloc.h>
24#include <stdio.h>
25#include <stdlib.h>
26#include <string.h>
27#include "../view_group.h"
28#include "../util.h"
29#include "../file_sorter.h"
30#include "util.h"
31#include "../mapreduce/mapreduce.h"
32
33#define BUF_SIZE 8192
34
35int main(int argc, char *argv[])
36{
37    view_group_info_t *group_info = NULL;
38    char buf[BUF_SIZE];
39    char **source_files = NULL;
40    char *tmp_dir = NULL;
41    int i;
42    int batch_size;
43    couchstore_error_t ret = (couchstore_error_t)2;
44    int is_sorted = 0;
45    view_group_update_stats_t stats;
46    sized_buf header_buf = {NULL, 0};
47    sized_buf header_outbuf = {NULL, 0};
48    view_error_t error_info = {NULL, NULL, "GENERIC"};
49    cb_thread_t exit_thread;
50
51    (void) argc;
52    (void) argv;
53
54    /*
55     * Disable buffering for stdout/stderr since index updater messages
56     * needs to be immediately available at erlang side
57     */
58    setvbuf(stdout, (char *) NULL, _IONBF, 0);
59    setvbuf(stderr, (char *) NULL, _IONBF, 0);
60
61    if (set_binary_mode() < 0) {
62        fprintf(stderr, "Error setting binary mode\n");
63        goto out;
64    }
65
66    /* Set all stats counters to zero */
67    memset((char *) &stats, 0, sizeof(view_group_update_stats_t));
68
69    if (couchstore_read_line(stdin, buf, BUF_SIZE) != buf) {
70        fprintf(stderr, "Error reading temporary directory path\n");
71        ret = COUCHSTORE_ERROR_INVALID_ARGUMENTS;
72        goto out;
73    }
74
75    tmp_dir = cb_strdup(buf);
76    if (tmp_dir == NULL) {
77        fprintf(stderr, "Memory allocation failure\n");
78        ret = COUCHSTORE_ERROR_ALLOC_FAIL;
79        goto out;
80    }
81
82    group_info = couchstore_read_view_group_info(stdin, stderr);
83    if (group_info == NULL) {
84        ret = COUCHSTORE_ERROR_ALLOC_FAIL;
85        goto out;
86    }
87
88    if (couchstore_read_line(stdin, buf, BUF_SIZE) != buf) {
89        fprintf(stderr, "Error reading is_sorted flag\n");
90        ret = COUCHSTORE_ERROR_INVALID_ARGUMENTS;
91        goto out;
92    }
93
94    if (!strncmp(buf, "s", 1)) {
95        is_sorted = 1;
96    }
97
98    source_files = (char **) cb_calloc(group_info->num_btrees + 1, sizeof(char *));
99    if (source_files == NULL) {
100        fprintf(stderr, "Memory allocation failure\n");
101        ret = COUCHSTORE_ERROR_ALLOC_FAIL;
102        goto out;
103    }
104
105    for (i = 0; i <= group_info->num_btrees; ++i) {
106        if (couchstore_read_line(stdin, buf, BUF_SIZE) != buf) {
107            if (i == 0) {
108                fprintf(stderr, "Error reading source file for id btree\n");
109            } else {
110                fprintf(stderr,
111                        "Error reading source file for btree %d\n", i - 1);
112            }
113            ret = COUCHSTORE_ERROR_INVALID_ARGUMENTS;
114            goto out;
115        }
116
117        source_files[i] = cb_strdup(buf);
118        if (source_files[i] == NULL) {
119            fprintf(stderr, "Memory allocation failure\n");
120            ret = COUCHSTORE_ERROR_ALLOC_FAIL;
121            goto out;
122        }
123    }
124
125    batch_size = couchstore_read_int(stdin, buf, sizeof(buf), &ret);
126    if (ret != COUCHSTORE_SUCCESS) {
127        fprintf(stderr, "Error reading batch size\n");
128        ret = COUCHSTORE_ERROR_INVALID_ARGUMENTS;
129        goto out;
130    }
131
132    header_buf.size = couchstore_read_int(stdin, buf, sizeof(buf), &ret);
133    if (ret != COUCHSTORE_SUCCESS) {
134        fprintf(stderr, "Error reading viewgroup header size\n");
135        goto out;
136    }
137
138    header_buf.buf = (char*)cb_malloc(header_buf.size);
139    if (header_buf.buf == NULL) {
140            fprintf(stderr, "Memory allocation failure\n");
141            ret = COUCHSTORE_ERROR_ALLOC_FAIL;
142            goto out;
143    }
144
145    if (fread(header_buf.buf, header_buf.size, 1, stdin) != 1) {
146        fprintf(stderr,
147                "Error reading viewgroup header from stdin\n");
148        goto out;
149    }
150
151    ret = (couchstore_error_t)start_exit_listener(&exit_thread, 1 /*uses_v8*/);
152    if (ret) {
153        fprintf(stderr, "Error starting stdin exit listener thread\n");
154        goto out;
155    }
156
157    mapreduce_init();
158    ret = couchstore_update_view_group(group_info,
159                                      source_files[0],
160                                      (const char **) &source_files[1],
161                                      batch_size,
162                                      &header_buf,
163                                      is_sorted,
164                                      tmp_dir,
165                                      &stats,
166                                      &header_outbuf,
167                                      &error_info);
168
169    mapreduce_deinit();
170
171    if (ret != COUCHSTORE_SUCCESS) {
172        if (error_info.error_msg != NULL && error_info.view_name != NULL) {
173            fprintf(stderr,
174                    "%s Error updating index for view `%s`, reason: %s\n",
175                    error_info.idx_type,
176                    error_info.view_name,
177                    error_info.error_msg);
178        }
179        goto out;
180    }
181
182    fprintf(stdout, "Header Len : %lu\n", header_outbuf.size);
183    fwrite(header_outbuf.buf, header_outbuf.size, 1, stdout);
184    fprintf(stdout, "\n");
185
186    fprintf(stdout,"Results ="
187                   " id_inserts : %" PRIu64
188                   ", id_deletes : %" PRIu64
189                   ", kv_inserts : %" PRIu64
190                   ", kv_deletes : %" PRIu64
191                   ", cleanups : %" PRIu64 "\n",
192                   stats.ids_inserted,
193                   stats.ids_removed,
194                   stats.kvs_inserted,
195                   stats.kvs_removed,
196                   stats.purged);
197
198out:
199    if (source_files != NULL) {
200        for (i = 0; i <= group_info->num_btrees; ++i) {
201            cb_free(source_files[i]);
202        }
203        cb_free(source_files);
204    }
205
206    couchstore_free_view_group_info(group_info);
207    cb_free((void *) error_info.error_msg);
208    cb_free((void *) error_info.view_name);
209    cb_free((void *) header_buf.buf);
210    cb_free((void *) header_outbuf.buf);
211    cb_free(tmp_dir);
212
213    int ret_int = (int)ret;
214    ret_int = (ret_int < 0) ? (100 + ret_int) : ret_int;
215    _exit(ret_int);
216}
217