1 /* -*- c-basic-offset: 2 -*- */ 2 /* 3 Copyright(C) 2009-2017 Brazil 4 5 This library is free software; you can redistribute it and/or 6 modify it under the terms of the GNU Lesser General Public 7 License version 2.1 as published by the Free Software Foundation. 8 9 This library is distributed in the hope that it will be useful, 10 but WITHOUT ANY WARRANTY; without even the implied warranty of 11 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 12 Lesser General Public License for more details. 13 14 You should have received a copy of the GNU Lesser General Public 15 License along with this library; if not, write to the Free Software 16 Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1335 USA 17 */ 18 19 #include <string.h> 20 #include <stdio.h> 21 #include <ctype.h> 22 #include <fcntl.h> 23 #include <sys/stat.h> 24 #include <errno.h> 25 26 #ifdef WIN32 27 # define GROONGA_MAIN 28 #endif /* WIN32 */ 29 #include <grn.h> 30 31 #include <grn_com.h> 32 #include <grn_ctx_impl.h> 33 #include <grn_proc.h> 34 #include <grn_db.h> 35 #include <grn_util.h> 36 #include <grn_error.h> 37 38 #ifdef HAVE_SYS_WAIT_H 39 # include <sys/wait.h> 40 #endif /* HAVE_SYS_WAIT_H */ 41 #ifdef HAVE_SYS_SOCKET_H 42 # include <sys/socket.h> 43 #endif /* HAVE_SYS_SOCKET_H */ 44 #ifndef WIN32 45 # include <netinet/in.h> 46 #endif /* WIN32 */ 47 48 #ifdef HAVE_SYS_RESOURCE_H 49 # include <sys/resource.h> 50 #endif /* HAVE_SYS_RESOURCE_H */ 51 52 #ifdef HAVE_SYS_SYSCTL_H 53 # include <sys/sysctl.h> 54 #endif /* HAVE_SYS_SYSCTL_H */ 55 56 #ifdef WIN32 57 # include <io.h> 58 # include <direct.h> 59 #else /* WIN32 */ 60 # include <sys/uio.h> 61 #endif /* WIN32 */ 62 63 #ifndef USE_MSG_NOSIGNAL 64 # ifdef MSG_NOSIGNAL 65 # undef MSG_NOSIGNAL 66 # endif 67 # define MSG_NOSIGNAL 0 68 #endif /* USE_MSG_NOSIGNAL */ 69 70 #ifndef STDIN_FILENO 71 # define STDIN_FILENO 0 72 #endif /* STDIN_FILENO */ 73 #ifndef STDOUT_FILENO 74 # define STDOUT_FILENO 1 75 #endif /* STDOUT_FILENO */ 76 #ifndef STDERR_FILENO 77 # define STDERR_FILENO 2 78 #endif /* STDERR_FILENO */ 79 80 #define DEFAULT_HTTP_PORT 10041 81 #define DEFAULT_GQTP_PORT 10043 82 #define DEFAULT_DEST "localhost" 83 #define DEFAULT_MAX_N_FLOATING_THREADS 8 84 #define MAX_CON 0x10000 85 86 #define RLIMIT_NOFILE_MINIMUM 4096 87 88 static char bind_address[HOST_NAME_MAX + 1]; 89 static char hostname[HOST_NAME_MAX + 1]; 90 static int port = DEFAULT_GQTP_PORT; 91 static int batchmode; 92 static int number_of_lines = 0; 93 static int newdb; 94 static grn_bool is_daemon_mode = GRN_FALSE; 95 static int (*do_client)(int argc, char **argv); 96 static int (*do_server)(char *path); 97 static const char *pid_file_path = NULL; 98 static const char *input_path = NULL; 99 static grn_file_reader *input_reader = NULL; 100 static FILE *output = NULL; 101 static grn_bool is_memcached_mode = GRN_FALSE; 102 static const char *memcached_column_name = NULL; 103 104 static int ready_notify_pipe[2]; 105 #define PIPE_READ 0 106 #define PIPE_WRITE 1 107 108 static grn_encoding encoding; 109 static const char *windows_event_source_name = "Groonga"; 110 static grn_bool use_windows_event_log = GRN_FALSE; 111 static grn_obj http_response_server_line; 112 113 static int 114 grn_rc_to_exit_code(grn_rc rc) 115 { 116 if (rc == GRN_SUCCESS) { 117 return EXIT_SUCCESS; 118 } else { 119 return EXIT_FAILURE; 120 } 121 } 122 123 static void 124 break_accept_event_loop(grn_ctx *ctx) 125 { 126 grn_com *client; 127 const char *address; 128 129 if (strcmp(bind_address, "0.0.0.0") == 0) { 130 address = "127.0.0.1"; 131 } else if (strcmp(bind_address, "::") == 0) { 132 address = "::1"; 133 } else { 134 address = bind_address; 135 } 136 client = grn_com_copen(ctx, NULL, address, port); 137 if (client) { 138 grn_com_close(ctx, client); 139 } 140 } 141 142 #ifdef GRN_WITH_LIBEDIT 143 #include <locale.h> 144 #include <histedit.h> 145 static EditLine *line_editor = NULL; 146 static HistoryW *line_editor_history = NULL; 147 static HistEventW line_editor_history_event; 148 static char line_editor_history_path[PATH_MAX] = ""; 149 150 static const wchar_t * 151 line_editor_prompt(EditLine *e __attribute__((unused))) 152 { 153 return L"> "; 154 } 155 static const wchar_t * const line_editor_editor = L"emacs"; 156 157 static void 158 line_editor_init(int argc __attribute__((unused)), char *argv[]) 159 { 160 const char * const HOME_PATH = getenv("HOME"); 161 const char * const HISTORY_PATH = "/.groonga-history"; 162 163 setlocale(LC_ALL, ""); 164 165 if (strlen(HOME_PATH) + strlen(HISTORY_PATH) < PATH_MAX) { 166 grn_strcpy(line_editor_history_path, PATH_MAX, HOME_PATH); 167 grn_strcat(line_editor_history_path, PATH_MAX, HISTORY_PATH); 168 } else { 169 line_editor_history_path[0] = '\0'; 170 } 171 172 line_editor_history = history_winit(); 173 history_w(line_editor_history, &line_editor_history_event, H_SETSIZE, 200); 174 if (line_editor_history_path[0]) { 175 history_w(line_editor_history, &line_editor_history_event, 176 H_LOAD, line_editor_history_path); 177 } 178 179 line_editor = el_init(argv[0], stdin, stdout, stderr); 180 el_wset(line_editor, EL_PROMPT, &line_editor_prompt); 181 el_wset(line_editor, EL_EDITOR, line_editor_editor); 182 el_wset(line_editor, EL_HIST, history_w, line_editor_history); 183 el_source(line_editor, NULL); 184 } 185 186 static void 187 line_editor_fin(void) 188 { 189 if (line_editor) { 190 el_end(line_editor); 191 if (line_editor_history) { 192 if (line_editor_history_path[0]) { 193 history_w(line_editor_history, &line_editor_history_event, 194 H_SAVE, line_editor_history_path); 195 } 196 history_wend(line_editor_history); 197 } 198 } 199 } 200 201 static grn_rc 202 line_editor_fgets(grn_ctx *ctx, grn_obj *buf) 203 { 204 grn_rc rc = GRN_SUCCESS; 205 const wchar_t *line; 206 int nchar; 207 line = el_wgets(line_editor, &nchar); 208 if (nchar > 0) { 209 int i; 210 char multibyte_buf[MB_CUR_MAX]; 211 size_t multibyte_len; 212 mbstate_t ps; 213 history_w(line_editor_history, &line_editor_history_event, H_ENTER, line); 214 memset(&ps, 0, sizeof(ps)); 215 wcrtomb(NULL, L'\0', &ps); 216 for (i = 0; i < nchar; i++) { 217 multibyte_len = wcrtomb(multibyte_buf, line[i], &ps); 218 if (multibyte_len == (size_t)-1) { 219 GRN_LOG(ctx, GRN_LOG_WARNING, 220 "[prompt][libedit] failed to read input: %s", strerror(errno)); 221 rc = GRN_INVALID_ARGUMENT; 222 } else { 223 GRN_TEXT_PUT(ctx, buf, multibyte_buf, multibyte_len); 224 } 225 } 226 } else { 227 rc = GRN_END_OF_DATA; 228 } 229 return rc; 230 } 231 #endif /* GRN_WITH_LIBEDIT */ 232 233 inline static grn_rc 234 read_next_line(grn_ctx *ctx, grn_obj *buf) 235 { 236 static int the_first_read = GRN_TRUE; 237 grn_rc rc = GRN_SUCCESS; 238 if (!batchmode) { 239 #ifdef GRN_WITH_LIBEDIT 240 rc = line_editor_fgets(ctx, buf); 241 #else 242 fprintf(stderr, "> "); 243 fflush(stderr); 244 rc = grn_file_reader_read_line(ctx, input_reader, buf); 245 #endif 246 } else { 247 rc = grn_file_reader_read_line(ctx, input_reader, buf); 248 if (rc != GRN_END_OF_DATA) { 249 number_of_lines++; 250 } 251 } 252 if (the_first_read && GRN_TEXT_LEN(buf) > 0) { 253 const char bom[] = {0xef, 0xbb, 0xbf}; 254 if (GRN_CTX_GET_ENCODING(ctx) == GRN_ENC_UTF8 && 255 GRN_TEXT_LEN(buf) > 3 && !memcmp(GRN_TEXT_VALUE(buf), bom, 3)) { 256 grn_obj buf_without_bom; 257 GRN_TEXT_INIT(&buf_without_bom, 0); 258 GRN_TEXT_PUT(ctx, &buf_without_bom, 259 GRN_TEXT_VALUE(buf) + 3, GRN_TEXT_LEN(buf) - 3); 260 GRN_TEXT_SET(ctx, buf, 261 GRN_TEXT_VALUE(&buf_without_bom), 262 GRN_TEXT_LEN(&buf_without_bom)); 263 grn_obj_unlink(ctx, &buf_without_bom); 264 } 265 the_first_read = GRN_FALSE; 266 } 267 if (GRN_TEXT_LEN(buf) > 0 && 268 GRN_TEXT_VALUE(buf)[GRN_TEXT_LEN(buf) - 1] == '\n') { 269 grn_bulk_truncate(ctx, buf, GRN_TEXT_LEN(buf) - 1); 270 } 271 if (GRN_TEXT_LEN(buf) > 0 && 272 GRN_TEXT_VALUE(buf)[GRN_TEXT_LEN(buf) - 1] == '\r') { 273 grn_bulk_truncate(ctx, buf, GRN_TEXT_LEN(buf) - 1); 274 } 275 return rc; 276 } 277 278 inline static grn_rc 279 prompt(grn_ctx *ctx, grn_obj *buf) 280 { 281 grn_rc rc = GRN_SUCCESS; 282 grn_bool need_next_line = GRN_TRUE; 283 GRN_BULK_REWIND(buf); 284 while (need_next_line) { 285 rc = read_next_line(ctx, buf); 286 if (rc == GRN_SUCCESS && 287 GRN_TEXT_LEN(buf) > 0 && 288 GRN_TEXT_VALUE(buf)[GRN_TEXT_LEN(buf) - 1] == '\\') { 289 grn_bulk_truncate(ctx, buf, GRN_TEXT_LEN(buf) - 1); 290 need_next_line = GRN_TRUE; 291 } else { 292 need_next_line = GRN_FALSE; 293 } 294 } 295 return rc; 296 } 297 298 static void 299 output_envelope(grn_ctx *ctx, grn_rc rc, grn_obj *head, grn_obj *body, grn_obj *foot) 300 { 301 grn_output_envelope(ctx, rc, head, body, foot, input_path, number_of_lines); 302 } 303 304 static void 305 s_output_raw(grn_ctx *ctx, int flags, FILE *stream) 306 { 307 char *chunk = NULL; 308 unsigned int chunk_size = 0; 309 int recv_flags; 310 311 grn_ctx_recv(ctx, &chunk, &chunk_size, &recv_flags); 312 if (chunk_size > 0) { 313 fwrite(chunk, 1, chunk_size, stream); 314 } 315 316 if (flags & GRN_CTX_TAIL) { 317 grn_obj *command; 318 319 if (grn_ctx_get_output_type(ctx) == GRN_CONTENT_GROONGA_COMMAND_LIST && 320 chunk_size > 0 && 321 chunk[chunk_size - 1] != '\n') { 322 fwrite("\n", 1, 1, stream); 323 } 324 fflush(stream); 325 326 command = GRN_CTX_USER_DATA(ctx)->ptr; 327 GRN_BULK_REWIND(command); 328 } 329 } 330 331 static void 332 s_output_typed(grn_ctx *ctx, int flags, FILE *stream) 333 { 334 if (ctx && ctx->impl && (flags & GRN_CTX_TAIL)) { 335 char *chunk = NULL; 336 unsigned int chunk_size = 0; 337 int recv_flags; 338 grn_obj body; 339 grn_obj *command; 340 341 GRN_TEXT_INIT(&body, 0); 342 grn_ctx_recv(ctx, &chunk, &chunk_size, &recv_flags); 343 GRN_TEXT_SET(ctx, &body, chunk, chunk_size); 344 345 if (GRN_TEXT_LEN(&body) || ctx->rc) { 346 grn_obj head, foot; 347 GRN_TEXT_INIT(&head, 0); 348 GRN_TEXT_INIT(&foot, 0); 349 output_envelope(ctx, ctx->rc, &head, &body, &foot); 350 fwrite(GRN_TEXT_VALUE(&head), 1, GRN_TEXT_LEN(&head), stream); 351 fwrite(GRN_TEXT_VALUE(&body), 1, GRN_TEXT_LEN(&body), stream); 352 fwrite(GRN_TEXT_VALUE(&foot), 1, GRN_TEXT_LEN(&foot), stream); 353 fputc('\n', stream); 354 fflush(stream); 355 GRN_OBJ_FIN(ctx, &head); 356 GRN_OBJ_FIN(ctx, &foot); 357 } 358 GRN_OBJ_FIN(ctx, &body); 359 360 command = GRN_CTX_USER_DATA(ctx)->ptr; 361 GRN_BULK_REWIND(command); 362 } 363 } 364 365 static void 366 s_output(grn_ctx *ctx, int flags, void *arg) 367 { 368 FILE *stream = (FILE *)arg; 369 370 switch (grn_ctx_get_output_type(ctx)) { 371 case GRN_CONTENT_GROONGA_COMMAND_LIST : 372 case GRN_CONTENT_NONE : 373 s_output_raw(ctx, flags, stream); 374 break; 375 default : 376 s_output_typed(ctx, flags, stream); 377 break; 378 } 379 } 380 381 static int 382 do_alone(int argc, char **argv) 383 { 384 int exit_code = EXIT_FAILURE; 385 char *path = NULL; 386 grn_obj *db; 387 grn_ctx ctx_, *ctx = &ctx_; 388 grn_ctx_init(ctx, 0); 389 if (argc > 0 && argv) { path = *argv++; argc--; } 390 db = (newdb || !path) ? grn_db_create(ctx, path, NULL) : grn_db_open(ctx, path); 391 if (db) { 392 grn_obj command; 393 GRN_TEXT_INIT(&command, 0); 394 GRN_CTX_USER_DATA(ctx)->ptr = &command; 395 grn_ctx_recv_handler_set(ctx, s_output, output); 396 if (!argc) { 397 grn_obj text; 398 GRN_TEXT_INIT(&text, 0); 399 while (prompt(ctx, &text) != GRN_END_OF_DATA) { 400 GRN_TEXT_PUT(ctx, &command, GRN_TEXT_VALUE(&text), GRN_TEXT_LEN(&text)); 401 grn_ctx_send(ctx, GRN_TEXT_VALUE(&text), GRN_TEXT_LEN(&text), 0); 402 if (ctx->stat == GRN_CTX_QUIT) { break; } 403 } 404 exit_code = grn_rc_to_exit_code(ctx->rc); 405 grn_obj_unlink(ctx, &text); 406 } else { 407 grn_rc rc; 408 rc = grn_ctx_sendv(ctx, argc, argv, 0); 409 exit_code = grn_rc_to_exit_code(rc); 410 } 411 grn_obj_unlink(ctx, &command); 412 grn_obj_close(ctx, db); 413 } else { 414 fprintf(stderr, "db open failed (%s): %s\n", path, ctx->errbuf); 415 } 416 grn_ctx_fin(ctx); 417 return exit_code; 418 } 419 420 static int 421 c_output(grn_ctx *ctx) 422 { 423 int flags; 424 char *str; 425 unsigned int str_len; 426 do { 427 grn_ctx_recv(ctx, &str, &str_len, &flags); 428 /* 429 if (ctx->rc) { 430 fprintf(stderr, "grn_ctx_recv failed\n"); 431 return -1; 432 } 433 */ 434 if (str_len || ctx->rc) { 435 grn_obj head, body, foot; 436 GRN_TEXT_INIT(&head, 0); 437 GRN_TEXT_INIT(&body, GRN_OBJ_DO_SHALLOW_COPY); 438 GRN_TEXT_INIT(&foot, 0); 439 if (ctx->rc == GRN_SUCCESS) { 440 GRN_TEXT_SET(ctx, &body, str, str_len); 441 } else { 442 ERR(ctx->rc, "%.*s", str_len, str); 443 } 444 output_envelope(ctx, ctx->rc, &head, &body, &foot); 445 fwrite(GRN_TEXT_VALUE(&head), 1, GRN_TEXT_LEN(&head), output); 446 fwrite(GRN_TEXT_VALUE(&body), 1, GRN_TEXT_LEN(&body), output); 447 fwrite(GRN_TEXT_VALUE(&foot), 1, GRN_TEXT_LEN(&foot), output); 448 fputc('\n', output); 449 fflush(output); 450 GRN_OBJ_FIN(ctx, &head); 451 GRN_OBJ_FIN(ctx, &body); 452 GRN_OBJ_FIN(ctx, &foot); 453 } 454 } while ((flags & GRN_CTX_MORE)); 455 return 0; 456 } 457 458 static int 459 g_client(int argc, char **argv) 460 { 461 int exit_code = EXIT_FAILURE; 462 grn_ctx ctx_, *ctx = &ctx_; 463 const char *hostname = DEFAULT_DEST; 464 if (argc > 0 && argv) { hostname = *argv++; argc--; } 465 grn_ctx_init(ctx, 0); 466 if (!grn_ctx_connect(ctx, hostname, port, 0)) { 467 if (!argc) { 468 grn_obj text; 469 GRN_TEXT_INIT(&text, 0); 470 while (prompt(ctx, &text) != GRN_END_OF_DATA) { 471 grn_ctx_send(ctx, GRN_TEXT_VALUE(&text), GRN_TEXT_LEN(&text), 0); 472 exit_code = grn_rc_to_exit_code(ctx->rc); 473 if (ctx->rc != GRN_SUCCESS) { break; } 474 if (c_output(ctx)) { goto exit; } 475 if (ctx->stat == GRN_CTX_QUIT) { break; } 476 } 477 grn_obj_unlink(ctx, &text); 478 } else { 479 grn_rc rc; 480 rc = grn_ctx_sendv(ctx, argc, argv, 0); 481 exit_code = grn_rc_to_exit_code(rc); 482 if (c_output(ctx)) { goto exit; } 483 } 484 } else { 485 fprintf(stderr, "grn_ctx_connect failed (%s:%d)\n", hostname, port); 486 } 487 exit : 488 grn_ctx_fin(ctx); 489 return exit_code; 490 } 491 492 /* server */ 493 494 typedef void (*grn_edge_dispatcher_func)(grn_ctx *ctx, grn_edge *edge); 495 typedef void (*grn_handler_func)(grn_ctx *ctx, grn_obj *msg); 496 497 static grn_com_queue ctx_new; 498 static grn_com_queue ctx_old; 499 static grn_mutex q_mutex; 500 static grn_cond q_cond; 501 static uint32_t n_running_threads = 0; 502 static uint32_t n_floating_threads = 0; 503 static uint32_t max_n_floating_threads; 504 505 static uint32_t 506 groonga_get_thread_limit(void *data) 507 { 508 return max_n_floating_threads; 509 } 510 511 static void 512 groonga_set_thread_limit(uint32_t new_limit, void *data) 513 { 514 uint32_t i; 515 uint32_t current_n_floating_threads; 516 static uint32_t n_changing_threads = 0; 517 uint32_t prev_n_changing_threads; 518 519 GRN_ATOMIC_ADD_EX(&n_changing_threads, 1, prev_n_changing_threads); 520 521 MUTEX_LOCK_ENSURE(&grn_gctx, q_mutex); 522 current_n_floating_threads = n_floating_threads; 523 max_n_floating_threads = new_limit; 524 MUTEX_UNLOCK(q_mutex); 525 526 if (prev_n_changing_threads > 0) { 527 GRN_ATOMIC_ADD_EX(&n_changing_threads, -1, prev_n_changing_threads); 528 return; 529 } 530 531 if (current_n_floating_threads > new_limit) { 532 for (i = 0; i < current_n_floating_threads; i++) { 533 MUTEX_LOCK_ENSURE(&grn_gctx, q_mutex); 534 COND_SIGNAL(q_cond); 535 MUTEX_UNLOCK(q_mutex); 536 } 537 } 538 539 while (GRN_TRUE) { 540 grn_bool is_reduced; 541 MUTEX_LOCK_ENSURE(&grn_gctx, q_mutex); 542 is_reduced = (n_running_threads <= max_n_floating_threads); 543 if (!is_reduced && n_floating_threads > 0) { 544 COND_SIGNAL(q_cond); 545 } 546 MUTEX_UNLOCK(q_mutex); 547 if (is_reduced) { 548 break; 549 } 550 grn_nanosleep(1000000); 551 } 552 553 GRN_ATOMIC_ADD_EX(&n_changing_threads, -1, prev_n_changing_threads); 554 } 555 556 typedef struct { 557 grn_mutex mutex; 558 grn_ctx ctx; 559 grn_pat *entries; 560 uint64_t earliest_unix_time_msec; 561 } request_timer_data; 562 static request_timer_data the_request_timer_data; 563 564 static void * 565 request_timer_register(const char *request_id, 566 unsigned int request_id_size, 567 double timeout, 568 void *user_data) 569 { 570 request_timer_data *data = user_data; 571 grn_id id = GRN_ID_NIL; 572 573 { 574 grn_ctx *ctx = &(data->ctx); 575 grn_bool is_first_timer; 576 grn_timeval tv; 577 uint64_t timeout_unix_time_msec; 578 void *value; 579 580 MUTEX_LOCK(data->mutex); 581 is_first_timer = (grn_pat_size(ctx, data->entries) == 0); 582 grn_timeval_now(ctx, &tv); 583 timeout_unix_time_msec = GRN_TIMEVAL_TO_MSEC(&tv) + (timeout * 1000); 584 while (GRN_TRUE) { 585 int added; 586 id = grn_pat_add(ctx, data->entries, 587 &timeout_unix_time_msec, sizeof(uint64_t), 588 &value, &added); 589 if (added != 0) { 590 break; 591 } 592 timeout_unix_time_msec++; 593 } 594 grn_memcpy(value, &request_id_size, sizeof(unsigned int)); 595 grn_memcpy(((uint8_t *)value) + sizeof(unsigned int), 596 request_id, request_id_size); 597 if (data->earliest_unix_time_msec == 0 || 598 data->earliest_unix_time_msec > timeout_unix_time_msec) { 599 data->earliest_unix_time_msec = timeout_unix_time_msec; 600 } 601 if (is_first_timer) { 602 break_accept_event_loop(ctx); 603 } 604 MUTEX_UNLOCK(data->mutex); 605 } 606 607 return (void *)(uint64_t)id; 608 } 609 610 static void 611 request_timer_unregister(void *timer_id, 612 void *user_data) 613 { 614 request_timer_data *data = user_data; 615 grn_id id = (grn_id)(uint64_t)timer_id; 616 617 { 618 grn_ctx *ctx = &(data->ctx); 619 uint64_t timeout_unix_time_msec; 620 int key_size; 621 622 MUTEX_LOCK(data->mutex); 623 key_size = grn_pat_get_key(ctx, 624 data->entries, 625 id, 626 &timeout_unix_time_msec, 627 sizeof(uint64_t)); 628 if (key_size > 0) { 629 grn_pat_delete_by_id(ctx, data->entries, id, NULL); 630 if (data->earliest_unix_time_msec >= timeout_unix_time_msec) { 631 data->earliest_unix_time_msec = 0; 632 } 633 } 634 MUTEX_UNLOCK(data->mutex); 635 } 636 } 637 638 static void 639 request_timer_fin(void *user_data) 640 { 641 request_timer_data *data = user_data; 642 643 { 644 grn_ctx *ctx = &(data->ctx); 645 grn_pat_close(ctx, data->entries); 646 grn_ctx_fin(ctx); 647 MUTEX_FIN(data->mutex); 648 } 649 } 650 651 static void 652 request_timer_init(void) 653 { 654 static grn_request_timer timer; 655 request_timer_data *data = &the_request_timer_data; 656 grn_ctx *ctx; 657 658 MUTEX_INIT(data->mutex); 659 ctx = &(data->ctx); 660 grn_ctx_init(ctx, 0); 661 data->entries = grn_pat_create(ctx, 662 NULL, 663 sizeof(uint64_t), 664 GRN_TABLE_MAX_KEY_SIZE, 665 GRN_OBJ_KEY_UINT); 666 data->earliest_unix_time_msec = 0; 667 668 timer.user_data = data; 669 timer.register_func = request_timer_register; 670 timer.unregister_func = request_timer_unregister; 671 timer.fin_func = request_timer_fin; 672 673 grn_request_timer_set(&timer); 674 } 675 676 static grn_bool 677 request_timer_ensure_earliest_unix_time_msec(void) 678 { 679 request_timer_data *data = &the_request_timer_data; 680 grn_ctx *ctx; 681 grn_pat_cursor *cursor; 682 683 if (data->earliest_unix_time_msec > 0) { 684 return GRN_TRUE; 685 } 686 687 ctx = &(data->ctx); 688 cursor = grn_pat_cursor_open(ctx, data->entries, 689 NULL, 0, 690 NULL, 0, 691 0, 1, GRN_CURSOR_ASCENDING); 692 if (!cursor) { 693 return GRN_FALSE; 694 } 695 while (grn_pat_cursor_next(ctx, cursor) != GRN_ID_NIL) { 696 void *key; 697 uint64_t timeout_unix_time_msec; 698 699 grn_pat_cursor_get_key(ctx, cursor, &key); 700 timeout_unix_time_msec = *(uint64_t *)key; 701 data->earliest_unix_time_msec = timeout_unix_time_msec; 702 break; 703 } 704 grn_pat_cursor_close(ctx, cursor); 705 706 return data->earliest_unix_time_msec > 0; 707 } 708 709 static int 710 request_timer_get_poll_timeout(void) 711 { 712 request_timer_data *data = &the_request_timer_data; 713 int timeout = 1000; 714 grn_ctx *ctx; 715 grn_timeval tv; 716 717 MUTEX_LOCK(data->mutex); 718 ctx = &(data->ctx); 719 if (grn_pat_size(ctx, data->entries) == 0) { 720 goto exit; 721 } 722 723 if (!request_timer_ensure_earliest_unix_time_msec()) { 724 goto exit; 725 } 726 727 grn_timeval_now(ctx, &tv); 728 timeout = data->earliest_unix_time_msec - GRN_TIMEVAL_TO_MSEC(&tv); 729 if (timeout < 0) { 730 timeout = 0; 731 } else if (timeout > 1000) { 732 timeout = 1000; 733 } 734 735 exit : 736 MUTEX_UNLOCK(data->mutex); 737 738 return timeout; 739 } 740 741 static void 742 request_timer_process_timeout(void) 743 { 744 request_timer_data *data = &the_request_timer_data; 745 grn_ctx *ctx; 746 grn_timeval tv; 747 uint64_t max; 748 grn_pat_cursor *cursor; 749 750 ctx = &(data->ctx); 751 if (grn_pat_size(ctx, data->entries) == 0) { 752 return; 753 } 754 755 grn_timeval_now(ctx, &tv); 756 max = GRN_TIMEVAL_TO_MSEC(&tv); 757 cursor = grn_pat_cursor_open(ctx, data->entries, 758 NULL, 0, 759 &max, sizeof(uint64_t), 760 0, -1, GRN_CURSOR_ASCENDING); 761 if (!cursor) { 762 return; 763 } 764 765 grn_id id; 766 while ((id = grn_pat_cursor_next(ctx, cursor)) != GRN_ID_NIL) { 767 void *value; 768 const char *request_id; 769 unsigned int request_id_size; 770 771 grn_pat_cursor_get_value(ctx, cursor, &value); 772 request_id_size = *((unsigned int *)value); 773 request_id = (const char *)(((uint8_t *)value) + sizeof(unsigned int)); 774 grn_request_canceler_cancel(request_id, request_id_size); 775 } 776 grn_pat_cursor_close(ctx, cursor); 777 } 778 779 static void 780 reset_ready_notify_pipe(void) 781 { 782 ready_notify_pipe[PIPE_READ] = 0; 783 ready_notify_pipe[PIPE_WRITE] = 0; 784 } 785 786 static void 787 close_ready_notify_pipe(void) 788 { 789 if (ready_notify_pipe[PIPE_READ] > 0) { 790 close(ready_notify_pipe[PIPE_READ]); 791 } 792 if (ready_notify_pipe[PIPE_WRITE] > 0) { 793 close(ready_notify_pipe[PIPE_WRITE]); 794 } 795 reset_ready_notify_pipe(); 796 } 797 798 static void 799 send_ready_notify(void) 800 { 801 if (ready_notify_pipe[PIPE_WRITE] > 0) { 802 const char *ready_notify_message = "ready"; 803 write(ready_notify_pipe[PIPE_WRITE], 804 ready_notify_message, 805 strlen(ready_notify_message)); 806 } 807 close_ready_notify_pipe(); 808 } 809 810 static void 811 create_pid_file(void) 812 { 813 FILE *pid_file = NULL; 814 815 if (!pid_file_path) { 816 return; 817 } 818 819 pid_file = fopen(pid_file_path, "w"); 820 if (!pid_file) { 821 fprintf(stderr, 822 "Failed to open PID file: <%s>: <%s>\n", 823 pid_file_path, grn_strerror(errno)); 824 return; 825 } 826 827 { 828 #ifdef WIN32 829 DWORD pid; 830 pid = GetCurrentProcessId(); 831 fprintf(pid_file, "%" GRN_FMT_DWORD "\n", pid); 832 #else /* WIN32 */ 833 pid_t pid; 834 pid = grn_getpid(); 835 fprintf(pid_file, "%d\n", pid); 836 #endif /* WIN32 */ 837 } 838 fclose(pid_file); 839 } 840 841 static void 842 clean_pid_file(void) 843 { 844 if (pid_file_path) { 845 grn_unlink(pid_file_path); 846 } 847 } 848 849 static int 850 daemonize(void) 851 { 852 int exit_code = EXIT_SUCCESS; 853 #ifndef WIN32 854 855 if (pipe(ready_notify_pipe) == -1) { 856 reset_ready_notify_pipe(); 857 } 858 859 switch (fork()) { 860 case 0: 861 break; 862 case -1: 863 perror("fork"); 864 return EXIT_FAILURE; 865 default: 866 wait(NULL); 867 if (ready_notify_pipe[PIPE_READ] > 0) { 868 int max_fd; 869 fd_set read_fds; 870 FD_ZERO(&read_fds); 871 FD_SET(ready_notify_pipe[PIPE_READ], &read_fds); 872 max_fd = ready_notify_pipe[PIPE_READ] + 1; 873 select(max_fd, &read_fds, NULL, NULL, NULL); 874 } 875 close_ready_notify_pipe(); 876 _exit(EXIT_SUCCESS); 877 } 878 switch (fork()) { 879 case 0: 880 if (pid_file_path) { 881 create_pid_file(); 882 } else { 883 pid_t pid; 884 pid = grn_getpid(); 885 fprintf(stderr, "%d\n", pid); 886 } 887 break; 888 case -1: 889 perror("fork"); 890 return EXIT_FAILURE; 891 default: 892 close_ready_notify_pipe(); 893 _exit(EXIT_SUCCESS); 894 } 895 { 896 int null_fd; 897 grn_open(null_fd, "/dev/null", O_RDWR); 898 if (null_fd != -1) { 899 dup2(null_fd, STDIN_FILENO); 900 dup2(null_fd, STDOUT_FILENO); 901 dup2(null_fd, STDERR_FILENO); 902 if (null_fd > STDERR_FILENO) { grn_close(null_fd); } 903 } 904 } 905 #endif /* WIN32 */ 906 return exit_code; 907 } 908 909 static void 910 run_server_loop(grn_ctx *ctx, grn_com_event *ev) 911 { 912 request_timer_init(); 913 while (!grn_com_event_poll(ctx, ev, request_timer_get_poll_timeout()) && 914 grn_gctx.stat != GRN_CTX_QUIT) { 915 grn_edge *edge; 916 while ((edge = (grn_edge *)grn_com_queue_deque(ctx, &ctx_old))) { 917 grn_obj *msg; 918 while ((msg = (grn_obj *)grn_com_queue_deque(ctx, &edge->send_old))) { 919 grn_msg_close(&edge->ctx, msg); 920 } 921 while ((msg = (grn_obj *)grn_com_queue_deque(ctx, &edge->recv_new))) { 922 grn_msg_close(ctx, msg); 923 } 924 grn_ctx_fin(&edge->ctx); 925 if (edge->com->has_sid && edge->com->opaque == edge) { 926 grn_com_close(ctx, edge->com); 927 } 928 grn_edges_delete(ctx, edge); 929 } 930 request_timer_process_timeout(); 931 /* todo : log stat */ 932 } 933 for (;;) { 934 MUTEX_LOCK_ENSURE(ctx, q_mutex); 935 if (n_running_threads == n_floating_threads) { break; } 936 MUTEX_UNLOCK(q_mutex); 937 grn_nanosleep(1000000); 938 } 939 { 940 grn_edge *edge; 941 GRN_HASH_EACH(ctx, grn_edges, id, NULL, NULL, &edge, { 942 grn_obj *obj; 943 while ((obj = (grn_obj *)grn_com_queue_deque(ctx, &edge->send_old))) { 944 grn_msg_close(&edge->ctx, obj); 945 } 946 while ((obj = (grn_obj *)grn_com_queue_deque(ctx, &edge->recv_new))) { 947 grn_msg_close(ctx, obj); 948 } 949 grn_ctx_fin(&edge->ctx); 950 if (edge->com->has_sid) { 951 grn_com_close(ctx, edge->com); 952 } 953 grn_edges_delete(ctx, edge); 954 }); 955 } 956 { 957 grn_com *com; 958 GRN_HASH_EACH(ctx, ev->hash, id, NULL, NULL, &com, { grn_com_close(ctx, com); }); 959 } 960 } 961 962 static int 963 run_server(grn_ctx *ctx, grn_obj *db, grn_com_event *ev, 964 grn_edge_dispatcher_func dispatcher, grn_handler_func handler) 965 { 966 int exit_code = EXIT_SUCCESS; 967 struct hostent *he; 968 if (!(he = gethostbyname(hostname))) { 969 send_ready_notify(); 970 SOERR("gethostbyname"); 971 } else { 972 ev->opaque = db; 973 grn_edges_init(ctx, dispatcher); 974 if (!grn_com_sopen(ctx, ev, bind_address, port, handler, he)) { 975 send_ready_notify(); 976 run_server_loop(ctx, ev); 977 exit_code = EXIT_SUCCESS; 978 } else { 979 send_ready_notify(); 980 fprintf(stderr, "grn_com_sopen failed (%s:%d): %s\n", 981 bind_address, port, ctx->errbuf); 982 } 983 grn_edges_fin(ctx); 984 } 985 return exit_code; 986 } 987 988 static grn_bool memcached_init(grn_ctx *ctx); 989 990 static int 991 start_service(grn_ctx *ctx, const char *db_path, 992 grn_edge_dispatcher_func dispatcher, grn_handler_func handler) 993 { 994 int exit_code = EXIT_SUCCESS; 995 grn_com_event ev; 996 997 if (is_daemon_mode) { 998 exit_code = daemonize(); 999 if (exit_code != EXIT_SUCCESS) { 1000 return exit_code; 1001 } 1002 } else { 1003 create_pid_file(); 1004 } 1005 1006 if (!grn_com_event_init(ctx, &ev, MAX_CON, sizeof(grn_com))) { 1007 grn_obj *db; 1008 db = (newdb || !db_path) ? grn_db_create(ctx, db_path, NULL) : grn_db_open(ctx, db_path); 1009 if (db) { 1010 if (is_memcached_mode) { 1011 if (!memcached_init(ctx)) { 1012 fprintf(stderr, "failed to initialize memcached mode: %s\n", 1013 ctx->errbuf); 1014 exit_code = EXIT_FAILURE; 1015 send_ready_notify(); 1016 } 1017 } 1018 if (exit_code == EXIT_SUCCESS) { 1019 exit_code = run_server(ctx, db, &ev, dispatcher, handler); 1020 } 1021 grn_obj_close(ctx, db); 1022 } else { 1023 fprintf(stderr, "db open failed (%s): %s\n", db_path, ctx->errbuf); 1024 exit_code = EXIT_FAILURE; 1025 send_ready_notify(); 1026 } 1027 grn_com_event_fin(ctx, &ev); 1028 } else { 1029 fprintf(stderr, "grn_com_event_init failed\n"); 1030 exit_code = EXIT_FAILURE; 1031 send_ready_notify(); 1032 } 1033 1034 clean_pid_file(); 1035 1036 return exit_code; 1037 } 1038 1039 typedef struct { 1040 grn_msg *msg; 1041 grn_bool in_body; 1042 grn_bool is_chunked; 1043 } ht_context; 1044 1045 static void 1046 h_output_set_header(grn_ctx *ctx, 1047 grn_obj *header, 1048 grn_rc rc, 1049 long long int content_length, 1050 grn_obj *foot) 1051 { 1052 switch (rc) { 1053 case GRN_SUCCESS : 1054 GRN_TEXT_SETS(ctx, header, "HTTP/1.1 200 OK\r\n"); 1055 break; 1056 case GRN_INVALID_ARGUMENT : 1057 case GRN_FUNCTION_NOT_IMPLEMENTED : 1058 case GRN_SYNTAX_ERROR : 1059 GRN_TEXT_SETS(ctx, header, "HTTP/1.1 400 Bad Request\r\n"); 1060 break; 1061 case GRN_NO_SUCH_FILE_OR_DIRECTORY : 1062 GRN_TEXT_SETS(ctx, header, "HTTP/1.1 404 Not Found\r\n"); 1063 break; 1064 case GRN_CANCEL : 1065 GRN_TEXT_SETS(ctx, header, "HTTP/1.1 408 Request Timeout\r\n"); 1066 break; 1067 default : 1068 GRN_TEXT_SETS(ctx, header, "HTTP/1.1 500 Internal Server Error\r\n"); 1069 break; 1070 } 1071 GRN_TEXT_PUT(ctx, header, 1072 GRN_TEXT_VALUE(&http_response_server_line), 1073 GRN_TEXT_LEN(&http_response_server_line)); 1074 GRN_TEXT_PUTS(ctx, header, "Content-Type: "); 1075 if (grn_ctx_get_output_type(ctx) == GRN_CONTENT_JSON && 1076 foot && 1077 GRN_TEXT_LEN(foot) > 0 && 1078 GRN_TEXT_VALUE(foot)[GRN_TEXT_LEN(foot) - 1] == ';') { 1079 GRN_TEXT_PUTS(ctx, header, "application/javascript"); 1080 } else { 1081 GRN_TEXT_PUTS(ctx, header, grn_ctx_get_mime_type(ctx)); 1082 } 1083 GRN_TEXT_PUTS(ctx, header, "\r\n"); 1084 if (content_length >= 0) { 1085 GRN_TEXT_PUTS(ctx, header, "Connection: close\r\n"); 1086 GRN_TEXT_PUTS(ctx, header, "Content-Length: "); 1087 grn_text_lltoa(ctx, header, content_length); 1088 GRN_TEXT_PUTS(ctx, header, "\r\n"); 1089 } else { 1090 GRN_TEXT_PUTS(ctx, header, "Transfer-Encoding: chunked\r\n"); 1091 } 1092 GRN_TEXT_PUTS(ctx, header, "\r\n"); 1093 } 1094 1095 static void 1096 h_output_send(grn_ctx *ctx, grn_sock fd, 1097 grn_obj *header, grn_obj *head, grn_obj *body, grn_obj *foot) 1098 { 1099 ssize_t ret; 1100 ssize_t len = 0; 1101 #ifdef WIN32 1102 int n_buffers = 0; 1103 WSABUF wsabufs[4]; 1104 if (header) { 1105 wsabufs[n_buffers].buf = GRN_TEXT_VALUE(header); 1106 wsabufs[n_buffers].len = GRN_TEXT_LEN(header); 1107 len += GRN_TEXT_LEN(header); 1108 n_buffers++; 1109 } 1110 if (head) { 1111 wsabufs[n_buffers].buf = GRN_TEXT_VALUE(head); 1112 wsabufs[n_buffers].len = GRN_TEXT_LEN(head); 1113 len += GRN_TEXT_LEN(head); 1114 n_buffers++; 1115 } 1116 if (body) { 1117 wsabufs[n_buffers].buf = GRN_TEXT_VALUE(body); 1118 wsabufs[n_buffers].len = GRN_TEXT_LEN(body); 1119 len += GRN_TEXT_LEN(body); 1120 n_buffers++; 1121 } 1122 if (foot) { 1123 wsabufs[n_buffers].buf = GRN_TEXT_VALUE(foot); 1124 wsabufs[n_buffers].len = GRN_TEXT_LEN(foot); 1125 len += GRN_TEXT_LEN(foot); 1126 n_buffers++; 1127 } 1128 { 1129 DWORD sent; 1130 if (WSASend(fd, wsabufs, n_buffers, &sent, 0, NULL, NULL) == SOCKET_ERROR) { 1131 SOERR("WSASend"); 1132 } 1133 ret = sent; 1134 } 1135 #else /* WIN32 */ 1136 struct iovec msg_iov[4]; 1137 struct msghdr msg; 1138 msg.msg_name = NULL; 1139 msg.msg_namelen = 0; 1140 msg.msg_iov = msg_iov; 1141 msg.msg_iovlen = 0; 1142 msg.msg_control = NULL; 1143 msg.msg_controllen = 0; 1144 msg.msg_flags = 0; 1145 1146 if (header) { 1147 msg_iov[msg.msg_iovlen].iov_base = GRN_TEXT_VALUE(header); 1148 msg_iov[msg.msg_iovlen].iov_len = GRN_TEXT_LEN(header); 1149 len += GRN_TEXT_LEN(header); 1150 msg.msg_iovlen++; 1151 } 1152 if (head) { 1153 msg_iov[msg.msg_iovlen].iov_base = GRN_TEXT_VALUE(head); 1154 msg_iov[msg.msg_iovlen].iov_len = GRN_TEXT_LEN(head); 1155 len += GRN_TEXT_LEN(head); 1156 msg.msg_iovlen++; 1157 } 1158 if (body) { 1159 msg_iov[msg.msg_iovlen].iov_base = GRN_TEXT_VALUE(body); 1160 msg_iov[msg.msg_iovlen].iov_len = GRN_TEXT_LEN(body); 1161 len += GRN_TEXT_LEN(body); 1162 msg.msg_iovlen++; 1163 } 1164 if (foot) { 1165 msg_iov[msg.msg_iovlen].iov_base = GRN_TEXT_VALUE(foot); 1166 msg_iov[msg.msg_iovlen].iov_len = GRN_TEXT_LEN(foot); 1167 len += GRN_TEXT_LEN(foot); 1168 msg.msg_iovlen++; 1169 } 1170 if ((ret = sendmsg(fd, &msg, MSG_NOSIGNAL)) == -1) { 1171 SOERR("sendmsg"); 1172 } 1173 #endif /* WIN32 */ 1174 if (ret != len) { 1175 GRN_LOG(&grn_gctx, GRN_LOG_NOTICE, 1176 "couldn't send all data (%" GRN_FMT_LLD "/%" GRN_FMT_LLD ")", 1177 (long long int)ret, (long long int)len); 1178 } 1179 } 1180 1181 static void 1182 h_output_raw(grn_ctx *ctx, int flags, ht_context *hc) 1183 { 1184 grn_rc expr_rc = ctx->rc; 1185 grn_sock fd = hc->msg->u.fd; 1186 grn_obj header_; 1187 grn_obj head_; 1188 grn_obj body_; 1189 grn_obj foot_; 1190 grn_obj *header = NULL; 1191 grn_obj *head = NULL; 1192 grn_obj *body = NULL; 1193 grn_obj *foot = NULL; 1194 char *chunk = NULL; 1195 unsigned int chunk_size = 0; 1196 int recv_flags; 1197 grn_bool is_last_message = (flags & GRN_CTX_TAIL); 1198 1199 GRN_TEXT_INIT(&header_, 0); 1200 GRN_TEXT_INIT(&head_, 0); 1201 GRN_TEXT_INIT(&body_, GRN_OBJ_DO_SHALLOW_COPY); 1202 GRN_TEXT_INIT(&foot_, 0); 1203 1204 grn_ctx_recv(ctx, &chunk, &chunk_size, &recv_flags); 1205 GRN_TEXT_SET(ctx, &body_, chunk, chunk_size); 1206 1207 if (!hc->in_body) { 1208 if (is_last_message) { 1209 h_output_set_header(ctx, &header_, expr_rc, GRN_TEXT_LEN(&body_), NULL); 1210 hc->is_chunked = GRN_FALSE; 1211 } else { 1212 h_output_set_header(ctx, &header_, expr_rc, -1, NULL); 1213 hc->is_chunked = GRN_TRUE; 1214 } 1215 header = &header_; 1216 hc->in_body = GRN_TRUE; 1217 } 1218 1219 if (GRN_TEXT_LEN(&body_) > 0) { 1220 if (hc->is_chunked) { 1221 grn_text_printf(ctx, &head_, 1222 "%x\r\n", (unsigned int)GRN_TEXT_LEN(&body_)); 1223 head = &head_; 1224 GRN_TEXT_PUTS(ctx, &foot_, "\r\n"); 1225 foot = &foot_; 1226 } 1227 body = &body_; 1228 } 1229 1230 if (is_last_message) { 1231 if (hc->is_chunked) { 1232 GRN_TEXT_PUTS(ctx, &foot_, "0\r\n"); 1233 GRN_TEXT_PUTS(ctx, &foot_, "Connection: close\r\n"); 1234 GRN_TEXT_PUTS(ctx, &foot_, "\r\n"); 1235 foot = &foot_; 1236 } 1237 } 1238 1239 h_output_send(ctx, fd, header, head, body, foot); 1240 1241 GRN_OBJ_FIN(ctx, &foot_); 1242 GRN_OBJ_FIN(ctx, &body_); 1243 GRN_OBJ_FIN(ctx, &head_); 1244 GRN_OBJ_FIN(ctx, &header_); 1245 } 1246 1247 static void 1248 h_output_typed(grn_ctx *ctx, int flags, ht_context *hc) 1249 { 1250 grn_rc expr_rc = ctx->rc; 1251 grn_sock fd = hc->msg->u.fd; 1252 grn_obj header, head, body, foot; 1253 char *chunk = NULL; 1254 unsigned int chunk_size = 0; 1255 int recv_flags; 1256 grn_bool should_return_body; 1257 1258 if (!(flags & GRN_CTX_TAIL)) { return; } 1259 1260 switch (hc->msg->header.qtype) { 1261 case 'G' : 1262 case 'P' : 1263 should_return_body = GRN_TRUE; 1264 break; 1265 default : 1266 should_return_body = GRN_FALSE; 1267 break; 1268 } 1269 1270 GRN_TEXT_INIT(&header, 0); 1271 GRN_TEXT_INIT(&head, 0); 1272 GRN_TEXT_INIT(&body, 0); 1273 GRN_TEXT_INIT(&foot, 0); 1274 1275 grn_ctx_recv(ctx, &chunk, &chunk_size, &recv_flags); 1276 GRN_TEXT_SET(ctx, &body, chunk, chunk_size); 1277 1278 output_envelope(ctx, expr_rc, &head, &body, &foot); 1279 h_output_set_header(ctx, &header, expr_rc, 1280 GRN_TEXT_LEN(&head) + 1281 GRN_TEXT_LEN(&body) + 1282 GRN_TEXT_LEN(&foot), 1283 &foot); 1284 if (should_return_body) { 1285 h_output_send(ctx, fd, &header, &head, &body, &foot); 1286 } else { 1287 h_output_send(ctx, fd, &header, NULL, NULL, NULL); 1288 } 1289 GRN_OBJ_FIN(ctx, &foot); 1290 GRN_OBJ_FIN(ctx, &body); 1291 GRN_OBJ_FIN(ctx, &head); 1292 GRN_OBJ_FIN(ctx, &header); 1293 } 1294 1295 static void 1296 h_output(grn_ctx *ctx, int flags, void *arg) 1297 { 1298 ht_context *hc = (ht_context *)arg; 1299 1300 switch (grn_ctx_get_output_type(ctx)) { 1301 case GRN_CONTENT_GROONGA_COMMAND_LIST : 1302 case GRN_CONTENT_NONE : 1303 h_output_raw(ctx, flags, hc); 1304 break; 1305 default : 1306 h_output_typed(ctx, flags, hc); 1307 break; 1308 } 1309 } 1310 1311 static void 1312 do_htreq_get(grn_ctx *ctx, ht_context *hc) 1313 { 1314 grn_msg *msg = hc->msg; 1315 char *path = NULL; 1316 char *pathe = GRN_BULK_HEAD((grn_obj *)msg); 1317 char *e = GRN_BULK_CURR((grn_obj *)msg); 1318 for (;; pathe++) { 1319 if (e <= pathe + 6) { 1320 /* invalid request */ 1321 return; 1322 } 1323 if (*pathe == ' ') { 1324 if (!path) { 1325 path = pathe + 1; 1326 } else { 1327 if (!memcmp(pathe + 1, "HTTP/1", 6)) { 1328 break; 1329 } 1330 } 1331 } 1332 } 1333 grn_ctx_send(ctx, path, pathe - path, GRN_CTX_TAIL); 1334 } 1335 1336 typedef struct { 1337 const char *path_start; 1338 int path_length; 1339 long long int content_length; 1340 grn_bool have_100_continue; 1341 const char *body_start; 1342 } h_post_header; 1343 1344 #define STRING_EQUAL(string, string_length, constant_string)\ 1345 (string_length == strlen(constant_string) &&\ 1346 strncmp(string, constant_string, string_length) == 0) 1347 1348 #define STRING_EQUAL_CI(string, string_length, constant_string)\ 1349 (string_length == strlen(constant_string) &&\ 1350 grn_strncasecmp(string, constant_string, string_length) == 0) 1351 1352 static const char * 1353 do_htreq_post_parse_header_request_line(grn_ctx *ctx, 1354 const char *start, 1355 const char *end, 1356 h_post_header *header) 1357 { 1358 const char *current; 1359 1360 { 1361 const char *method = start; 1362 int method_length = -1; 1363 1364 for (current = method; current < end; current++) { 1365 if (current[0] == '\n') { 1366 return NULL; 1367 } 1368 if (current[0] == ' ') { 1369 method_length = current - method; 1370 current++; 1371 break; 1372 } 1373 } 1374 if (method_length == -1) { 1375 return NULL; 1376 } 1377 if (!STRING_EQUAL_CI(method, method_length, "POST")) { 1378 return NULL; 1379 } 1380 } 1381 1382 { 1383 header->path_start = current; 1384 header->path_length = -1; 1385 for (; current < end; current++) { 1386 if (current[0] == '\n') { 1387 return NULL; 1388 } 1389 if (current[0] == ' ') { 1390 header->path_length = current - header->path_start; 1391 current++; 1392 break; 1393 } 1394 } 1395 if (header->path_length == -1) { 1396 return NULL; 1397 } 1398 } 1399 1400 { 1401 const char *http_version_start = current; 1402 int http_version_length = -1; 1403 for (; current < end; current++) { 1404 if (current[0] == '\n') { 1405 http_version_length = current - http_version_start; 1406 if (http_version_length > 0 && 1407 http_version_start[http_version_length - 1] == '\r') { 1408 http_version_length--; 1409 } 1410 current++; 1411 break; 1412 } 1413 } 1414 if (http_version_length == -1) { 1415 return NULL; 1416 } 1417 if (!(STRING_EQUAL_CI(http_version_start, http_version_length, "HTTP/1.0") || 1418 STRING_EQUAL_CI(http_version_start, http_version_length, "HTTP/1.1"))) { 1419 return NULL; 1420 } 1421 } 1422 1423 return current; 1424 } 1425 1426 static const char * 1427 do_htreq_post_parse_header_values(grn_ctx *ctx, 1428 const char *start, 1429 const char *end, 1430 h_post_header *header) 1431 { 1432 const char *current; 1433 const char *name = start; 1434 int name_length = -1; 1435 const char *value = NULL; 1436 int value_length = -1; 1437 1438 for (current = start; current < end; current++) { 1439 switch (current[0]) { 1440 case '\n' : 1441 if (name_length == -1) { 1442 if (current - name == 1 && current[-1] == '\r') { 1443 return current + 1; 1444 } else { 1445 /* No ":" header line. TODO: report error. */ 1446 return NULL; 1447 } 1448 } else { 1449 while (value < current && value[0] == ' ') { 1450 value++; 1451 } 1452 value_length = current - value; 1453 if (value_length > 0 && value[value_length - 1] == '\r') { 1454 value_length--; 1455 } 1456 if (STRING_EQUAL_CI(name, name_length, "Content-Length")) { 1457 const char *rest; 1458 header->content_length = grn_atoll(value, value + value_length, &rest); 1459 if (rest != value + value_length) { 1460 /* Invalid Content-Length value. TODO: report error. */ 1461 header->content_length = -1; 1462 } 1463 } else if (STRING_EQUAL_CI(name, name_length, "Expect")) { 1464 if (STRING_EQUAL(value, value_length, "100-continue")) { 1465 header->have_100_continue = GRN_TRUE; 1466 } 1467 } 1468 } 1469 name = current + 1; 1470 name_length = -1; 1471 value = NULL; 1472 value_length = -1; 1473 break; 1474 case ':' : 1475 if (name_length == -1) { 1476 name_length = current - name; 1477 value = current + 1; 1478 } 1479 break; 1480 default : 1481 break; 1482 } 1483 } 1484 1485 return NULL; 1486 } 1487 1488 static grn_bool 1489 do_htreq_post_parse_header(grn_ctx *ctx, 1490 const char *start, 1491 const char *end, 1492 h_post_header *header) 1493 { 1494 const char *current; 1495 1496 current = do_htreq_post_parse_header_request_line(ctx, start, end, header); 1497 if (!current) { 1498 return GRN_FALSE; 1499 } 1500 current = do_htreq_post_parse_header_values(ctx, current, end, header); 1501 if (!current) { 1502 return GRN_FALSE; 1503 } 1504 1505 if (current == end) { 1506 header->body_start = NULL; 1507 } else { 1508 header->body_start = current; 1509 } 1510 1511 return GRN_TRUE; 1512 } 1513 1514 static void 1515 do_htreq_post(grn_ctx *ctx, ht_context *hc) 1516 { 1517 grn_msg *msg = hc->msg; 1518 grn_sock fd = msg->u.fd; 1519 const char *end; 1520 h_post_header header; 1521 1522 header.path_start = NULL; 1523 header.path_length = -1; 1524 header.content_length = -1; 1525 header.body_start = NULL; 1526 header.have_100_continue = GRN_FALSE; 1527 1528 end = GRN_BULK_CURR((grn_obj *)msg); 1529 if (!do_htreq_post_parse_header(ctx, 1530 GRN_BULK_HEAD((grn_obj *)msg), 1531 end, 1532 &header)) { 1533 return; 1534 } 1535 1536 grn_ctx_send(ctx, header.path_start, header.path_length, GRN_CTX_MORE); 1537 if (ctx->rc != GRN_SUCCESS) { 1538 ht_context context; 1539 context.msg = msg; 1540 context.in_body = GRN_FALSE; 1541 context.is_chunked = GRN_FALSE; 1542 h_output(ctx, GRN_CTX_TAIL, &context); 1543 return; 1544 } 1545 1546 if (header.have_100_continue) { 1547 const char *continue_message = "HTTP/1.1 100 Continue\r\n"; 1548 ssize_t send_size; 1549 int send_flags = MSG_NOSIGNAL; 1550 send_size = send(fd, continue_message, strlen(continue_message), send_flags); 1551 if (send_size == -1) { 1552 SOERR("send"); 1553 return; 1554 } 1555 } 1556 1557 { 1558 grn_obj chunk_buffer; 1559 long long int read_content_length = 0; 1560 1561 GRN_TEXT_INIT(&chunk_buffer, 0); 1562 while (read_content_length < header.content_length) { 1563 #define POST_BUFFER_SIZE 8192 1564 char buffer[POST_BUFFER_SIZE]; 1565 const char *buffer_start, *buffer_current, *buffer_end; 1566 1567 if (header.body_start) { 1568 buffer_start = header.body_start; 1569 buffer_end = end; 1570 header.body_start = NULL; 1571 } else { 1572 ssize_t recv_length; 1573 int recv_flags = 0; 1574 recv_length = recv(fd, buffer, POST_BUFFER_SIZE, recv_flags); 1575 if (recv_length == 0) { 1576 break; 1577 } 1578 if (recv_length == -1) { 1579 SOERR("recv"); 1580 break; 1581 } 1582 buffer_start = buffer; 1583 buffer_end = buffer_start + recv_length; 1584 } 1585 read_content_length += buffer_end - buffer_start; 1586 1587 buffer_current = buffer_end - 1; 1588 for (; buffer_current > buffer_start; buffer_current--) { 1589 grn_bool is_separator; 1590 switch (buffer_current[0]) { 1591 case '\n' : 1592 case ',' : 1593 is_separator = GRN_TRUE; 1594 break; 1595 default : 1596 is_separator = GRN_FALSE; 1597 break; 1598 } 1599 if (!is_separator) { 1600 continue; 1601 } 1602 1603 GRN_TEXT_PUT(ctx, 1604 &chunk_buffer, 1605 buffer_start, 1606 buffer_current + 1 - buffer_start); 1607 { 1608 int flags = 0; 1609 if (!(read_content_length == header.content_length && 1610 buffer_current + 1 == buffer_end)) { 1611 flags |= GRN_CTX_MORE; 1612 } else { 1613 flags |= GRN_CTX_TAIL; 1614 } 1615 grn_ctx_send(ctx, 1616 GRN_TEXT_VALUE(&chunk_buffer), 1617 GRN_TEXT_LEN(&chunk_buffer), 1618 flags); 1619 } 1620 buffer_start = buffer_current + 1; 1621 GRN_BULK_REWIND(&chunk_buffer); 1622 break; 1623 } 1624 if (buffer_end > buffer_start) { 1625 GRN_TEXT_PUT(ctx, &chunk_buffer, 1626 buffer_start, buffer_end - buffer_start); 1627 } 1628 #undef POST_BUFFER_SIZE 1629 1630 if (ctx->rc != GRN_SUCCESS) { 1631 break; 1632 } 1633 } 1634 1635 if (ctx->rc == GRN_CANCEL) { 1636 h_output(ctx, GRN_CTX_TAIL, hc); 1637 } else if (ctx->rc == GRN_SUCCESS && GRN_TEXT_LEN(&chunk_buffer) > 0) { 1638 grn_ctx_send(ctx, 1639 GRN_TEXT_VALUE(&chunk_buffer), 1640 GRN_TEXT_LEN(&chunk_buffer), 1641 GRN_CTX_TAIL); 1642 } 1643 1644 GRN_OBJ_FIN(ctx, &chunk_buffer); 1645 } 1646 } 1647 1648 static void 1649 do_htreq(grn_ctx *ctx, ht_context *hc) 1650 { 1651 grn_msg *msg = hc->msg; 1652 grn_com_header *header = &msg->header; 1653 switch (header->qtype) { 1654 case 'G' : /* GET */ 1655 case 'H' : /* HEAD */ 1656 do_htreq_get(ctx, hc); 1657 break; 1658 case 'P' : /* POST */ 1659 do_htreq_post(ctx, hc); 1660 break; 1661 } 1662 /* if (ctx->rc != GRN_OPERATION_WOULD_BLOCK) {...} */ 1663 grn_msg_close(ctx, (grn_obj *)msg); 1664 /* if not keep alive connection */ 1665 grn_sock_close(msg->u.fd); 1666 grn_com_event_start_accept(ctx, msg->acceptor->ev); 1667 } 1668 1669 enum { 1670 MBRES_SUCCESS = 0x00, 1671 MBRES_KEY_ENOENT = 0x01, 1672 MBRES_KEY_EEXISTS = 0x02, 1673 MBRES_E2BIG = 0x03, 1674 MBRES_EINVAL = 0x04, 1675 MBRES_NOT_STORED = 0x05, 1676 MBRES_UNKNOWN_COMMAND = 0x81, 1677 MBRES_ENOMEM = 0x82, 1678 }; 1679 1680 enum { 1681 MBCMD_GET = 0x00, 1682 MBCMD_SET = 0x01, 1683 MBCMD_ADD = 0x02, 1684 MBCMD_REPLACE = 0x03, 1685 MBCMD_DELETE = 0x04, 1686 MBCMD_INCREMENT = 0x05, 1687 MBCMD_DECREMENT = 0x06, 1688 MBCMD_QUIT = 0x07, 1689 MBCMD_FLUSH = 0x08, 1690 MBCMD_GETQ = 0x09, 1691 MBCMD_NOOP = 0x0a, 1692 MBCMD_VERSION = 0x0b, 1693 MBCMD_GETK = 0x0c, 1694 MBCMD_GETKQ = 0x0d, 1695 MBCMD_APPEND = 0x0e, 1696 MBCMD_PREPEND = 0x0f, 1697 MBCMD_STAT = 0x10, 1698 MBCMD_SETQ = 0x11, 1699 MBCMD_ADDQ = 0x12, 1700 MBCMD_REPLACEQ = 0x13, 1701 MBCMD_DELETEQ = 0x14, 1702 MBCMD_INCREMENTQ = 0x15, 1703 MBCMD_DECREMENTQ = 0x16, 1704 MBCMD_QUITQ = 0x17, 1705 MBCMD_FLUSHQ = 0x18, 1706 MBCMD_APPENDQ = 0x19, 1707 MBCMD_PREPENDQ = 0x1a 1708 }; 1709 1710 static grn_obj *cache_table = NULL; 1711 static grn_obj *cache_value = NULL; 1712 static grn_obj *cache_flags = NULL; 1713 static grn_obj *cache_expire = NULL; 1714 static grn_obj *cache_cas = NULL; 1715 1716 #define CTX_GET(name) (grn_ctx_get(ctx, (name), strlen(name))) 1717 1718 static grn_bool 1719 memcached_setup_flags_column(grn_ctx *ctx, const char *name) 1720 { 1721 cache_flags = grn_obj_column(ctx, cache_table, name, strlen(name)); 1722 if (cache_flags) { 1723 return GRN_TRUE; 1724 } 1725 1726 cache_flags = grn_column_create(ctx, cache_table, name, strlen(name), NULL, 1727 GRN_OBJ_COLUMN_SCALAR|GRN_OBJ_PERSISTENT, 1728 grn_ctx_at(ctx, GRN_DB_UINT32)); 1729 if (!cache_flags) { 1730 return GRN_FALSE; 1731 } 1732 1733 return GRN_TRUE; 1734 } 1735 1736 static grn_bool 1737 memcached_setup_expire_column(grn_ctx *ctx, const char *name) 1738 { 1739 cache_expire = grn_obj_column(ctx, cache_table, name, strlen(name)); 1740 if (cache_expire) { 1741 return GRN_TRUE; 1742 } 1743 1744 cache_expire = grn_column_create(ctx, cache_table, name, strlen(name), NULL, 1745 GRN_OBJ_COLUMN_SCALAR|GRN_OBJ_PERSISTENT, 1746 grn_ctx_at(ctx, GRN_DB_UINT32)); 1747 if (!cache_expire) { 1748 return GRN_FALSE; 1749 } 1750 1751 return GRN_TRUE; 1752 } 1753 1754 static grn_bool 1755 memcached_setup_cas_column(grn_ctx *ctx, const char *name) 1756 { 1757 cache_cas = grn_obj_column(ctx, cache_table, name, strlen(name)); 1758 if (cache_cas) { 1759 return GRN_TRUE; 1760 } 1761 1762 cache_cas = grn_column_create(ctx, cache_table, name, strlen(name), NULL, 1763 GRN_OBJ_COLUMN_SCALAR|GRN_OBJ_PERSISTENT, 1764 grn_ctx_at(ctx, GRN_DB_UINT64)); 1765 if (!cache_cas) { 1766 return GRN_FALSE; 1767 } 1768 1769 return GRN_TRUE; 1770 } 1771 1772 static grn_bool 1773 memcached_init(grn_ctx *ctx) 1774 { 1775 if (memcached_column_name) { 1776 cache_value = CTX_GET(memcached_column_name); 1777 if (!cache_value) { 1778 ERR(GRN_INVALID_ARGUMENT, 1779 "memcached column doesn't exist: <%s>", 1780 memcached_column_name); 1781 return GRN_FALSE; 1782 } 1783 if (!(grn_obj_is_column(ctx, cache_value) && 1784 ((cache_value->header.flags & GRN_OBJ_COLUMN_TYPE_MASK) == 1785 GRN_OBJ_COLUMN_SCALAR))) { 1786 grn_obj inspected; 1787 GRN_TEXT_INIT(&inspected, 0); 1788 grn_inspect(ctx, &inspected, cache_value); 1789 ERR(GRN_INVALID_ARGUMENT, 1790 "memcached column must be scalar column: <%.*s>", 1791 (int)GRN_TEXT_LEN(&inspected), 1792 GRN_TEXT_VALUE(&inspected)); 1793 GRN_OBJ_FIN(ctx, &inspected); 1794 return GRN_FALSE; 1795 } 1796 if (!(GRN_DB_SHORT_TEXT <= grn_obj_get_range(ctx, cache_value) && 1797 grn_obj_get_range(ctx, cache_value) <= GRN_DB_LONG_TEXT)) { 1798 grn_obj inspected; 1799 GRN_TEXT_INIT(&inspected, 0); 1800 grn_inspect(ctx, &inspected, cache_value); 1801 ERR(GRN_INVALID_ARGUMENT, 1802 "memcached column must be text column: <%.*s>", 1803 (int)GRN_TEXT_LEN(&inspected), 1804 GRN_TEXT_VALUE(&inspected)); 1805 GRN_OBJ_FIN(ctx, &inspected); 1806 return GRN_FALSE; 1807 } 1808 1809 cache_table = grn_ctx_at(ctx, cache_value->header.domain); 1810 if (cache_table->header.type == GRN_TABLE_NO_KEY) { 1811 grn_obj inspected; 1812 GRN_TEXT_INIT(&inspected, 0); 1813 grn_inspect(ctx, &inspected, cache_table); 1814 ERR(GRN_INVALID_ARGUMENT, 1815 "memcached column's table must be HASH_KEY, PAT_KEY or DAT_KEY table: " 1816 "<%.*s>", 1817 (int)GRN_TEXT_LEN(&inspected), 1818 GRN_TEXT_VALUE(&inspected)); 1819 GRN_OBJ_FIN(ctx, &inspected); 1820 return GRN_FALSE; 1821 } 1822 1823 { 1824 char column_name[GRN_TABLE_MAX_KEY_SIZE]; 1825 char value_column_name[GRN_TABLE_MAX_KEY_SIZE]; 1826 int value_column_name_size; 1827 1828 value_column_name_size = grn_column_name(ctx, cache_value, 1829 value_column_name, 1830 GRN_TABLE_MAX_KEY_SIZE); 1831 grn_snprintf(column_name, 1832 GRN_TABLE_MAX_KEY_SIZE, 1833 GRN_TABLE_MAX_KEY_SIZE, 1834 "%.*s_memcached_flags", 1835 value_column_name_size, 1836 value_column_name); 1837 if (!memcached_setup_flags_column(ctx, column_name)) { 1838 return GRN_FALSE; 1839 } 1840 grn_snprintf(column_name, 1841 GRN_TABLE_MAX_KEY_SIZE, 1842 GRN_TABLE_MAX_KEY_SIZE, 1843 "%.*s_memcached_expire", 1844 value_column_name_size, 1845 value_column_name); 1846 if (!memcached_setup_expire_column(ctx, column_name)) { 1847 return GRN_FALSE; 1848 } 1849 grn_snprintf(column_name, 1850 GRN_TABLE_MAX_KEY_SIZE, 1851 GRN_TABLE_MAX_KEY_SIZE, 1852 "%.*s_memcached_cas", 1853 value_column_name_size, 1854 value_column_name); 1855 if (!memcached_setup_cas_column(ctx, column_name)) { 1856 return GRN_FALSE; 1857 } 1858 } 1859 } else { 1860 const char *table_name = "Memcache"; 1861 const char *value_column_name = "value"; 1862 1863 cache_table = CTX_GET(table_name); 1864 if (!cache_table) { 1865 cache_table = grn_table_create(ctx, table_name, strlen(table_name), NULL, 1866 GRN_OBJ_TABLE_PAT_KEY|GRN_OBJ_PERSISTENT, 1867 grn_ctx_at(ctx, GRN_DB_SHORT_TEXT), 1868 NULL); 1869 if (!cache_table) { 1870 return GRN_FALSE; 1871 } 1872 } 1873 1874 cache_value = grn_obj_column(ctx, cache_table, 1875 value_column_name, 1876 strlen(value_column_name)); 1877 if (!cache_value) { 1878 cache_value = grn_column_create(ctx, cache_table, 1879 value_column_name, 1880 strlen(value_column_name), 1881 NULL, 1882 GRN_OBJ_COLUMN_SCALAR|GRN_OBJ_PERSISTENT, 1883 grn_ctx_at(ctx, GRN_DB_SHORT_TEXT)); 1884 if (!cache_value) { 1885 return GRN_FALSE; 1886 } 1887 } 1888 1889 if (!memcached_setup_flags_column(ctx, "flags")) { 1890 return GRN_FALSE; 1891 } 1892 if (!memcached_setup_expire_column(ctx, "expire")) { 1893 return GRN_FALSE; 1894 } 1895 if (!memcached_setup_cas_column(ctx, "cas")) { 1896 return GRN_FALSE; 1897 } 1898 } 1899 1900 return GRN_TRUE; 1901 } 1902 1903 #define RELATIVE_TIME_THRESH 1000000000 1904 1905 #define MBRES(ctx,re,status,key_len,extra_len,flags) do {\ 1906 grn_msg_set_property((ctx), (re), (status), (key_len), (extra_len));\ 1907 grn_msg_send((ctx), (re), (flags));\ 1908 } while (0) 1909 1910 #define GRN_MSG_MBRES(block) do {\ 1911 if (!quiet) {\ 1912 grn_obj *re = grn_msg_open_for_reply(ctx, (grn_obj *)msg, &edge->send_old);\ 1913 ((grn_msg *)re)->header.qtype = header->qtype;\ 1914 block\ 1915 }\ 1916 } while (0) 1917 1918 static uint64_t 1919 get_mbreq_cas_id() 1920 { 1921 static uint64_t cas_id = 0; 1922 /* FIXME: use GRN_ATOMIC_ADD_EX_64, but it is not implemented */ 1923 return ++cas_id; 1924 } 1925 1926 static void 1927 do_mbreq(grn_ctx *ctx, grn_edge *edge) 1928 { 1929 int quiet = 0; 1930 int flags = 0; 1931 grn_msg *msg = edge->msg; 1932 grn_com_header *header = &msg->header; 1933 1934 switch (header->qtype) { 1935 case MBCMD_GETQ : 1936 flags = GRN_CTX_MORE; 1937 /* fallthru */ 1938 case MBCMD_GET : 1939 { 1940 grn_id rid; 1941 uint16_t keylen = ntohs(header->keylen); 1942 char *key = GRN_BULK_HEAD((grn_obj *)msg); 1943 rid = grn_table_get(ctx, cache_table, key, keylen); 1944 if (!rid) { 1945 GRN_MSG_MBRES({ 1946 MBRES(ctx, re, MBRES_KEY_ENOENT, 0, 0, 0); 1947 }); 1948 } else { 1949 grn_timeval tv; 1950 uint32_t expire; 1951 { 1952 grn_obj expire_buf; 1953 GRN_UINT32_INIT(&expire_buf, 0); 1954 grn_obj_get_value(ctx, cache_expire, rid, &expire_buf); 1955 expire = GRN_UINT32_VALUE(&expire_buf); 1956 grn_obj_close(ctx, &expire_buf); 1957 } 1958 grn_timeval_now(ctx, &tv); 1959 if (expire && expire < tv.tv_sec) { 1960 grn_table_delete_by_id(ctx, cache_table, rid); 1961 GRN_MSG_MBRES({ 1962 MBRES(ctx, re, MBRES_KEY_ENOENT, 0, 0, 0); 1963 }); 1964 } else { 1965 grn_obj cas_buf; 1966 GRN_UINT64_INIT(&cas_buf, 0); 1967 grn_obj_get_value(ctx, cache_cas, rid, &cas_buf); 1968 GRN_MSG_MBRES({ 1969 grn_obj_get_value(ctx, cache_flags, rid, re); 1970 grn_obj_get_value(ctx, cache_value, rid, re); 1971 ((grn_msg *)re)->header.cas = GRN_UINT64_VALUE(&cas_buf); 1972 MBRES(ctx, re, MBRES_SUCCESS, 0, 4, flags); 1973 }); 1974 grn_obj_close(ctx, &cas_buf); 1975 } 1976 } 1977 } 1978 break; 1979 case MBCMD_SETQ : 1980 case MBCMD_ADDQ : 1981 case MBCMD_REPLACEQ : 1982 quiet = 1; 1983 /* fallthru */ 1984 case MBCMD_SET : 1985 case MBCMD_ADD : 1986 case MBCMD_REPLACE : 1987 { 1988 grn_id rid; 1989 uint32_t size = ntohl(header->size); 1990 uint16_t keylen = ntohs(header->keylen); 1991 uint8_t extralen = header->level; 1992 char *body = GRN_BULK_HEAD((grn_obj *)msg); 1993 uint32_t flags = *((uint32_t *)body); 1994 uint32_t expire = ntohl(*((uint32_t *)(body + 4))); 1995 uint32_t valuelen = size - keylen - extralen; 1996 char *key = body + 8; 1997 char *value = key + keylen; 1998 int added = 0; 1999 int f = (header->qtype == MBCMD_REPLACE || 2000 header->qtype == MBCMD_REPLACEQ) ? 0 : GRN_TABLE_ADD; 2001 GRN_ASSERT(extralen == 8); 2002 if (header->qtype == MBCMD_REPLACE || header->qtype == MBCMD_REPLACEQ) { 2003 rid = grn_table_get(ctx, cache_table, key, keylen); 2004 } else { 2005 rid = grn_table_add(ctx, cache_table, key, keylen, &added); 2006 } 2007 if (!rid) { 2008 GRN_MSG_MBRES({ 2009 MBRES(ctx, re, (f & GRN_TABLE_ADD) ? MBRES_ENOMEM : MBRES_NOT_STORED, 0, 0, 0); 2010 }); 2011 } else { 2012 if (added) { 2013 if (header->cas) { 2014 GRN_MSG_MBRES({ 2015 MBRES(ctx, re, MBRES_EINVAL, 0, 0, 0); 2016 }); 2017 } else { 2018 grn_obj text_buf, uint32_buf; 2019 GRN_TEXT_INIT(&text_buf, GRN_OBJ_DO_SHALLOW_COPY); 2020 GRN_TEXT_SET_REF(&text_buf, value, valuelen); 2021 grn_obj_set_value(ctx, cache_value, rid, &text_buf, GRN_OBJ_SET); 2022 GRN_UINT32_INIT(&uint32_buf, 0); 2023 GRN_UINT32_SET(ctx, &uint32_buf, flags); 2024 grn_obj_set_value(ctx, cache_flags, rid, &uint32_buf, GRN_OBJ_SET); 2025 if (expire && expire < RELATIVE_TIME_THRESH) { 2026 grn_timeval tv; 2027 grn_timeval_now(ctx, &tv); 2028 expire += tv.tv_sec; 2029 } 2030 GRN_UINT32_SET(ctx, &uint32_buf, expire); 2031 grn_obj_set_value(ctx, cache_expire, rid, &uint32_buf, GRN_OBJ_SET); 2032 grn_obj_close(ctx, &uint32_buf); 2033 { 2034 grn_obj cas_buf; 2035 uint64_t cas_id = get_mbreq_cas_id(); 2036 GRN_UINT64_INIT(&cas_buf, 0); 2037 GRN_UINT64_SET(ctx, &cas_buf, cas_id); 2038 grn_obj_set_value(ctx, cache_cas, rid, &cas_buf, GRN_OBJ_SET); 2039 grn_obj_close(ctx, &cas_buf); 2040 GRN_MSG_MBRES({ 2041 ((grn_msg *)re)->header.cas = cas_id; 2042 MBRES(ctx, re, MBRES_SUCCESS, 0, 0, 0); 2043 }); 2044 } 2045 } 2046 } else { 2047 if (header->qtype != MBCMD_SET && header->qtype != MBCMD_SETQ) { 2048 grn_obj uint32_buf; 2049 grn_timeval tv; 2050 uint32_t oexpire; 2051 2052 GRN_UINT32_INIT(&uint32_buf, 0); 2053 grn_obj_get_value(ctx, cache_expire, rid, &uint32_buf); 2054 oexpire = GRN_UINT32_VALUE(&uint32_buf); 2055 grn_timeval_now(ctx, &tv); 2056 2057 if (oexpire && oexpire < tv.tv_sec) { 2058 if (header->qtype == MBCMD_REPLACE || 2059 header->qtype == MBCMD_REPLACEQ) { 2060 grn_table_delete_by_id(ctx, cache_table, rid); 2061 GRN_MSG_MBRES({ 2062 MBRES(ctx, re, MBRES_NOT_STORED, 0, 0, 0); 2063 }); 2064 break; 2065 } 2066 } else if (header->qtype == MBCMD_ADD || 2067 header->qtype == MBCMD_ADDQ) { 2068 GRN_MSG_MBRES({ 2069 MBRES(ctx, re, MBRES_NOT_STORED, 0, 0, 0); 2070 }); 2071 break; 2072 } 2073 } 2074 { 2075 if (header->cas) { 2076 grn_obj cas_buf; 2077 GRN_UINT64_INIT(&cas_buf, 0); 2078 grn_obj_get_value(ctx, cache_cas, rid, &cas_buf); 2079 if (header->cas != GRN_UINT64_VALUE(&cas_buf)) { 2080 GRN_MSG_MBRES({ 2081 MBRES(ctx, re, MBRES_NOT_STORED, 0, 0, 0); 2082 }); 2083 } 2084 } 2085 { 2086 grn_obj text_buf, uint32_buf; 2087 GRN_TEXT_INIT(&text_buf, GRN_OBJ_DO_SHALLOW_COPY); 2088 GRN_TEXT_SET_REF(&text_buf, value, valuelen); 2089 grn_obj_set_value(ctx, cache_value, rid, &text_buf, GRN_OBJ_SET); 2090 GRN_UINT32_INIT(&uint32_buf, 0); 2091 GRN_UINT32_SET(ctx, &uint32_buf, flags); 2092 grn_obj_set_value(ctx, cache_flags, rid, &uint32_buf, GRN_OBJ_SET); 2093 if (expire && expire < RELATIVE_TIME_THRESH) { 2094 grn_timeval tv; 2095 grn_timeval_now(ctx, &tv); 2096 expire += tv.tv_sec; 2097 } 2098 GRN_UINT32_SET(ctx, &uint32_buf, expire); 2099 grn_obj_set_value(ctx, cache_expire, rid, &uint32_buf, GRN_OBJ_SET); 2100 { 2101 grn_obj cas_buf; 2102 uint64_t cas_id = get_mbreq_cas_id(); 2103 GRN_UINT64_INIT(&cas_buf, 0); 2104 GRN_UINT64_SET(ctx, &cas_buf, cas_id); 2105 grn_obj_set_value(ctx, cache_cas, rid, &cas_buf, GRN_OBJ_SET); 2106 GRN_MSG_MBRES({ 2107 ((grn_msg *)re)->header.cas = cas_id; 2108 MBRES(ctx, re, MBRES_SUCCESS, 0, 0, 0); 2109 }); 2110 } 2111 } 2112 } 2113 } 2114 } 2115 } 2116 break; 2117 case MBCMD_DELETEQ : 2118 quiet = 1; 2119 /* fallthru */ 2120 case MBCMD_DELETE : 2121 { 2122 grn_id rid; 2123 uint16_t keylen = ntohs(header->keylen); 2124 char *key = GRN_BULK_HEAD((grn_obj *)msg); 2125 rid = grn_table_get(ctx, cache_table, key, keylen); 2126 if (!rid) { 2127 /* GRN_LOG(ctx, GRN_LOG_NOTICE, "GET k=%d not found", keylen); */ 2128 GRN_MSG_MBRES({ 2129 MBRES(ctx, re, MBRES_KEY_ENOENT, 0, 0, 0); 2130 }); 2131 } else { 2132 grn_table_delete_by_id(ctx, cache_table, rid); 2133 GRN_MSG_MBRES({ 2134 MBRES(ctx, re, MBRES_SUCCESS, 0, 4, 0); 2135 }); 2136 } 2137 } 2138 break; 2139 case MBCMD_INCREMENTQ : 2140 case MBCMD_DECREMENTQ : 2141 quiet = 1; 2142 /* fallthru */ 2143 case MBCMD_INCREMENT : 2144 case MBCMD_DECREMENT : 2145 { 2146 grn_id rid; 2147 int added = 0; 2148 uint64_t delta, init; 2149 uint16_t keylen = ntohs(header->keylen); 2150 char *body = GRN_BULK_HEAD((grn_obj *)msg); 2151 char *key = body + 20; 2152 uint32_t expire = ntohl(*((uint32_t *)(body + 16))); 2153 grn_ntoh(&delta, body, 8); 2154 grn_ntoh(&init, body + 8, 8); 2155 GRN_ASSERT(header->level == 20); /* extralen */ 2156 if (expire == 0xffffffff) { 2157 rid = grn_table_get(ctx, cache_table, key, keylen); 2158 } else { 2159 rid = grn_table_add(ctx, cache_table, key, keylen, &added); 2160 } 2161 if (!rid) { 2162 GRN_MSG_MBRES({ 2163 MBRES(ctx, re, MBRES_KEY_ENOENT, 0, 0, 0); 2164 }); 2165 } else { 2166 grn_obj uint32_buf, text_buf; 2167 GRN_UINT32_INIT(&uint32_buf, 0); 2168 GRN_TEXT_INIT(&text_buf, GRN_OBJ_DO_SHALLOW_COPY); 2169 if (added) { 2170 GRN_TEXT_SET_REF(&text_buf, &init, 8); 2171 grn_obj_set_value(ctx, cache_value, rid, &text_buf, GRN_OBJ_SET); 2172 GRN_UINT32_SET(ctx, &uint32_buf, 0); 2173 grn_obj_set_value(ctx, cache_flags, rid, &uint32_buf, GRN_OBJ_SET); 2174 } else { 2175 grn_timeval tv; 2176 uint32_t oexpire; 2177 2178 grn_obj_get_value(ctx, cache_expire, rid, &uint32_buf); 2179 oexpire = GRN_UINT32_VALUE(&uint32_buf); 2180 grn_timeval_now(ctx, &tv); 2181 2182 if (oexpire && oexpire < tv.tv_sec) { 2183 if (expire == 0xffffffffU) { 2184 GRN_MSG_MBRES({ 2185 MBRES(ctx, re, MBRES_KEY_ENOENT, 0, 0, 0); 2186 }); 2187 break; 2188 } else { 2189 GRN_TEXT_SET_REF(&text_buf, &init, 8); 2190 grn_obj_set_value(ctx, cache_value, rid, &text_buf, GRN_OBJ_SET); 2191 GRN_UINT32_SET(ctx, &uint32_buf, 0); 2192 grn_obj_set_value(ctx, cache_flags, rid, &uint32_buf, GRN_OBJ_SET); 2193 } 2194 } else { 2195 grn_obj uint64_buf; 2196 GRN_UINT64_INIT(&uint64_buf, 0); 2197 GRN_UINT64_SET(ctx, &uint64_buf, delta); 2198 grn_obj_set_value(ctx, cache_value, rid, &uint64_buf, 2199 header->qtype == MBCMD_INCREMENT || 2200 header->qtype == MBCMD_INCREMENTQ 2201 ? GRN_OBJ_INCR 2202 : GRN_OBJ_DECR); 2203 } 2204 } 2205 if (expire && expire < RELATIVE_TIME_THRESH) { 2206 grn_timeval tv; 2207 grn_timeval_now(ctx, &tv); 2208 expire += tv.tv_sec; 2209 } 2210 GRN_UINT32_SET(ctx, &uint32_buf, expire); 2211 grn_obj_set_value(ctx, cache_expire, rid, &uint32_buf, GRN_OBJ_SET); 2212 GRN_MSG_MBRES({ 2213 /* TODO: get_mbreq_cas_id() */ 2214 grn_obj_get_value(ctx, cache_value, rid, re); 2215 grn_hton(&delta, (uint64_t *)GRN_BULK_HEAD(re), 8); 2216 GRN_TEXT_SET(ctx, re, &delta, sizeof(uint64_t)); 2217 MBRES(ctx, re, MBRES_SUCCESS, 0, sizeof(uint64_t), 0); 2218 }); 2219 } 2220 } 2221 break; 2222 case MBCMD_FLUSHQ : 2223 quiet = 1; 2224 /* fallthru */ 2225 case MBCMD_FLUSH : 2226 { 2227 uint32_t expire; 2228 uint8_t extralen = header->level; 2229 if (extralen) { 2230 char *body = GRN_BULK_HEAD((grn_obj *)msg); 2231 GRN_ASSERT(extralen == 4); 2232 expire = ntohl(*((uint32_t *)(body))); 2233 if (expire < RELATIVE_TIME_THRESH) { 2234 grn_timeval tv; 2235 grn_timeval_now(ctx, &tv); 2236 if (expire) { 2237 expire += tv.tv_sec; 2238 } else { 2239 expire = tv.tv_sec - 1; 2240 } 2241 } 2242 } else { 2243 grn_timeval tv; 2244 grn_timeval_now(ctx, &tv); 2245 expire = tv.tv_sec - 1; 2246 } 2247 { 2248 grn_obj exp_buf; 2249 GRN_UINT32_INIT(&exp_buf, 0); 2250 GRN_UINT32_SET(ctx, &exp_buf, expire); 2251 GRN_TABLE_EACH(ctx, cache_table, 0, 0, rid, NULL, NULL, NULL, { 2252 grn_obj_set_value(ctx, cache_expire, rid, &exp_buf, GRN_OBJ_SET); 2253 }); 2254 GRN_MSG_MBRES({ 2255 MBRES(ctx, re, MBRES_SUCCESS, 0, 4, 0); 2256 }); 2257 grn_obj_close(ctx, &exp_buf); 2258 } 2259 } 2260 break; 2261 case MBCMD_NOOP : 2262 break; 2263 case MBCMD_VERSION : 2264 GRN_MSG_MBRES({ 2265 grn_bulk_write(ctx, re, PACKAGE_VERSION, strlen(PACKAGE_VERSION)); 2266 MBRES(ctx, re, MBRES_SUCCESS, 0, 0, 0); 2267 }); 2268 break; 2269 case MBCMD_GETKQ : 2270 flags = GRN_CTX_MORE; 2271 /* fallthru */ 2272 case MBCMD_GETK : 2273 { 2274 grn_id rid; 2275 uint16_t keylen = ntohs(header->keylen); 2276 char *key = GRN_BULK_HEAD((grn_obj *)msg); 2277 rid = grn_table_get(ctx, cache_table, key, keylen); 2278 if (!rid) { 2279 GRN_MSG_MBRES({ 2280 MBRES(ctx, re, MBRES_KEY_ENOENT, 0, 0, 0); 2281 }); 2282 } else { 2283 grn_obj uint32_buf; 2284 grn_timeval tv; 2285 uint32_t expire; 2286 GRN_UINT32_INIT(&uint32_buf, 0); 2287 grn_obj_get_value(ctx, cache_expire, rid, &uint32_buf); 2288 expire = GRN_UINT32_VALUE(&uint32_buf); 2289 grn_timeval_now(ctx, &tv); 2290 if (expire && expire < tv.tv_sec) { 2291 grn_table_delete_by_id(ctx, cache_table, rid); 2292 GRN_MSG_MBRES({ 2293 MBRES(ctx, re, MBRES_KEY_ENOENT, 0, 0, 0); 2294 }); 2295 } else { 2296 grn_obj uint64_buf; 2297 GRN_UINT64_INIT(&uint64_buf, 0); 2298 grn_obj_get_value(ctx, cache_cas, rid, &uint64_buf); 2299 GRN_MSG_MBRES({ 2300 grn_obj_get_value(ctx, cache_flags, rid, re); 2301 grn_bulk_write(ctx, re, key, keylen); 2302 grn_obj_get_value(ctx, cache_value, rid, re); 2303 ((grn_msg *)re)->header.cas = GRN_UINT64_VALUE(&uint64_buf); 2304 MBRES(ctx, re, MBRES_SUCCESS, keylen, 4, flags); 2305 }); 2306 } 2307 } 2308 } 2309 break; 2310 case MBCMD_APPENDQ : 2311 case MBCMD_PREPENDQ : 2312 quiet = 1; 2313 /* fallthru */ 2314 case MBCMD_APPEND : 2315 case MBCMD_PREPEND : 2316 { 2317 grn_id rid; 2318 uint32_t size = ntohl(header->size); 2319 uint16_t keylen = ntohs(header->keylen); 2320 char *key = GRN_BULK_HEAD((grn_obj *)msg); 2321 char *value = key + keylen; 2322 uint32_t valuelen = size - keylen; 2323 rid = grn_table_add(ctx, cache_table, key, keylen, NULL); 2324 if (!rid) { 2325 GRN_MSG_MBRES({ 2326 MBRES(ctx, re, MBRES_ENOMEM, 0, 0, 0); 2327 }); 2328 } else { 2329 /* FIXME: check expire */ 2330 grn_obj buf; 2331 int flags = header->qtype == MBCMD_APPEND ? GRN_OBJ_APPEND : GRN_OBJ_PREPEND; 2332 GRN_TEXT_INIT(&buf, GRN_OBJ_DO_SHALLOW_COPY); 2333 GRN_TEXT_SET_REF(&buf, value, valuelen); 2334 grn_obj_set_value(ctx, cache_value, rid, &buf, flags); 2335 GRN_MSG_MBRES({ 2336 MBRES(ctx, re, MBRES_SUCCESS, 0, 0, 0); 2337 }); 2338 } 2339 } 2340 break; 2341 case MBCMD_STAT : 2342 { 2343 pid_t pid = grn_getpid(); 2344 GRN_MSG_MBRES({ 2345 grn_bulk_write(ctx, re, "pid", 3); 2346 grn_text_itoa(ctx, re, pid); 2347 MBRES(ctx, re, MBRES_SUCCESS, 3, 0, 0); 2348 }); 2349 } 2350 break; 2351 case MBCMD_QUITQ : 2352 quiet = 1; 2353 /* fallthru */ 2354 case MBCMD_QUIT : 2355 GRN_MSG_MBRES({ 2356 MBRES(ctx, re, MBRES_SUCCESS, 0, 0, 0); 2357 }); 2358 /* fallthru */ 2359 default : 2360 ctx->stat = GRN_CTX_QUIT; 2361 break; 2362 } 2363 } 2364 2365 /* worker thread */ 2366 2367 enum { 2368 EDGE_IDLE = 0x00, 2369 EDGE_WAIT = 0x01, 2370 EDGE_DOING = 0x02, 2371 EDGE_ABORT = 0x03, 2372 }; 2373 2374 static void 2375 check_rlimit_nofile(grn_ctx *ctx) 2376 { 2377 #ifndef WIN32 2378 struct rlimit limit; 2379 limit.rlim_cur = 0; 2380 limit.rlim_max = 0; 2381 getrlimit(RLIMIT_NOFILE, &limit); 2382 if (limit.rlim_cur < RLIMIT_NOFILE_MINIMUM) { 2383 limit.rlim_cur = RLIMIT_NOFILE_MINIMUM; 2384 limit.rlim_max = RLIMIT_NOFILE_MINIMUM; 2385 setrlimit(RLIMIT_NOFILE, &limit); 2386 limit.rlim_cur = 0; 2387 limit.rlim_max = 0; 2388 getrlimit(RLIMIT_NOFILE, &limit); 2389 } 2390 GRN_LOG(ctx, GRN_LOG_NOTICE, 2391 "RLIMIT_NOFILE(%" GRN_FMT_LLD ",%" GRN_FMT_LLD ")", 2392 (long long int)limit.rlim_cur, (long long int)limit.rlim_max); 2393 #endif /* WIN32 */ 2394 } 2395 2396 static grn_thread_func_result CALLBACK 2397 h_worker(void *arg) 2398 { 2399 ht_context hc; 2400 grn_ctx ctx_, *ctx = &ctx_; 2401 grn_ctx_init(ctx, 0); 2402 grn_ctx_use(ctx, (grn_obj *)arg); 2403 grn_ctx_recv_handler_set(ctx, h_output, &hc); 2404 MUTEX_LOCK_ENSURE(ctx, q_mutex); 2405 GRN_LOG(&grn_gctx, GRN_LOG_NOTICE, "thread start (%d/%d)", 2406 n_floating_threads, n_running_threads); 2407 while (n_running_threads <= max_n_floating_threads && 2408 grn_gctx.stat != GRN_CTX_QUIT) { 2409 grn_obj *msg; 2410 if (ctx->rc == GRN_CANCEL) { 2411 ctx->rc = GRN_SUCCESS; 2412 } 2413 n_floating_threads++; 2414 while (!(msg = (grn_obj *)grn_com_queue_deque(&grn_gctx, &ctx_new))) { 2415 COND_WAIT(q_cond, q_mutex); 2416 if (grn_gctx.stat == GRN_CTX_QUIT) { 2417 n_floating_threads--; 2418 goto exit; 2419 } 2420 if (n_running_threads > max_n_floating_threads) { 2421 n_floating_threads--; 2422 goto exit; 2423 } 2424 } 2425 n_floating_threads--; 2426 MUTEX_UNLOCK(q_mutex); 2427 hc.msg = (grn_msg *)msg; 2428 hc.in_body = GRN_FALSE; 2429 hc.is_chunked = GRN_FALSE; 2430 do_htreq(ctx, &hc); 2431 MUTEX_LOCK_ENSURE(ctx, q_mutex); 2432 } 2433 exit : 2434 n_running_threads--; 2435 GRN_LOG(&grn_gctx, GRN_LOG_NOTICE, "thread end (%d/%d)", 2436 n_floating_threads, n_running_threads); 2437 if (grn_gctx.stat == GRN_CTX_QUIT) { 2438 break_accept_event_loop(ctx); 2439 } 2440 grn_ctx_fin(ctx); 2441 MUTEX_UNLOCK(q_mutex); 2442 return GRN_THREAD_FUNC_RETURN_VALUE; 2443 } 2444 2445 static void 2446 h_handler(grn_ctx *ctx, grn_obj *msg) 2447 { 2448 grn_com *com = ((grn_msg *)msg)->u.peer; 2449 if (ctx->rc) { 2450 grn_com_close(ctx, com); 2451 grn_msg_close(ctx, msg); 2452 } else { 2453 grn_sock fd = com->fd; 2454 void *arg = com->ev->opaque; 2455 /* if not keep alive connection */ 2456 grn_com_event_del(ctx, com->ev, fd); 2457 ((grn_msg *)msg)->u.fd = fd; 2458 MUTEX_LOCK_ENSURE(ctx, q_mutex); 2459 grn_com_queue_enque(ctx, &ctx_new, (grn_com_queue_entry *)msg); 2460 if (n_floating_threads == 0 && n_running_threads < max_n_floating_threads) { 2461 grn_thread thread; 2462 n_running_threads++; 2463 if (THREAD_CREATE(thread, h_worker, arg)) { 2464 n_running_threads--; 2465 SERR("pthread_create"); 2466 } 2467 } 2468 COND_SIGNAL(q_cond); 2469 MUTEX_UNLOCK(q_mutex); 2470 } 2471 } 2472 2473 static int 2474 h_server(char *path) 2475 { 2476 int exit_code = EXIT_FAILURE; 2477 grn_ctx ctx_, *ctx = &ctx_; 2478 grn_ctx_init(ctx, 0); 2479 GRN_COM_QUEUE_INIT(&ctx_new); 2480 GRN_COM_QUEUE_INIT(&ctx_old); 2481 check_rlimit_nofile(ctx); 2482 GRN_TEXT_INIT(&http_response_server_line, 0); 2483 grn_text_printf(ctx, 2484 &http_response_server_line, 2485 "Server: %s/%s\r\n", 2486 grn_get_package_label(), 2487 grn_get_version()); 2488 exit_code = start_service(ctx, path, NULL, h_handler); 2489 GRN_OBJ_FIN(ctx, &http_response_server_line); 2490 grn_ctx_fin(ctx); 2491 return exit_code; 2492 } 2493 2494 static grn_thread_func_result CALLBACK 2495 g_worker(void *arg) 2496 { 2497 MUTEX_LOCK_ENSURE(NULL, q_mutex); 2498 GRN_LOG(&grn_gctx, GRN_LOG_NOTICE, "thread start (%d/%d)", 2499 n_floating_threads, n_running_threads); 2500 while (n_running_threads <= max_n_floating_threads && 2501 grn_gctx.stat != GRN_CTX_QUIT) { 2502 grn_ctx *ctx; 2503 grn_edge *edge; 2504 n_floating_threads++; 2505 while (!(edge = (grn_edge *)grn_com_queue_deque(&grn_gctx, &ctx_new))) { 2506 COND_WAIT(q_cond, q_mutex); 2507 if (grn_gctx.stat == GRN_CTX_QUIT) { 2508 n_floating_threads--; 2509 goto exit; 2510 } 2511 if (n_running_threads > max_n_floating_threads) { 2512 n_floating_threads--; 2513 goto exit; 2514 } 2515 } 2516 ctx = &edge->ctx; 2517 n_floating_threads--; 2518 if (edge->stat == EDGE_DOING) { continue; } 2519 if (edge->stat == EDGE_WAIT) { 2520 edge->stat = EDGE_DOING; 2521 while (!GRN_COM_QUEUE_EMPTYP(&edge->recv_new)) { 2522 grn_obj *msg; 2523 MUTEX_UNLOCK(q_mutex); 2524 /* if (edge->flags == GRN_EDGE_WORKER) */ 2525 while (ctx->stat != GRN_CTX_QUIT && 2526 (edge->msg = (grn_msg *)grn_com_queue_deque(ctx, &edge->recv_new))) { 2527 grn_com_header *header = &edge->msg->header; 2528 msg = (grn_obj *)edge->msg; 2529 switch (header->proto) { 2530 case GRN_COM_PROTO_MBREQ : 2531 do_mbreq(ctx, edge); 2532 break; 2533 case GRN_COM_PROTO_GQTP : 2534 grn_ctx_send(ctx, GRN_BULK_HEAD(msg), GRN_BULK_VSIZE(msg), header->flags); 2535 ERRCLR(ctx); 2536 if (ctx->rc == GRN_CANCEL) { 2537 ctx->rc = GRN_SUCCESS; 2538 } 2539 break; 2540 default : 2541 ctx->stat = GRN_CTX_QUIT; 2542 break; 2543 } 2544 grn_msg_close(ctx, msg); 2545 } 2546 while ((msg = (grn_obj *)grn_com_queue_deque(ctx, &edge->send_old))) { 2547 grn_msg_close(ctx, msg); 2548 } 2549 MUTEX_LOCK_ENSURE(ctx, q_mutex); 2550 if (ctx->stat == GRN_CTX_QUIT || edge->stat == EDGE_ABORT) { break; } 2551 } 2552 } 2553 if (ctx->stat == GRN_CTX_QUIT || edge->stat == EDGE_ABORT) { 2554 grn_com_queue_enque(&grn_gctx, &ctx_old, (grn_com_queue_entry *)edge); 2555 edge->stat = EDGE_ABORT; 2556 } else { 2557 edge->stat = EDGE_IDLE; 2558 } 2559 }; 2560 exit : 2561 n_running_threads--; 2562 GRN_LOG(&grn_gctx, GRN_LOG_NOTICE, "thread end (%d/%d)", 2563 n_floating_threads, n_running_threads); 2564 MUTEX_UNLOCK(q_mutex); 2565 return GRN_THREAD_FUNC_RETURN_VALUE; 2566 } 2567 2568 static void 2569 g_dispatcher(grn_ctx *ctx, grn_edge *edge) 2570 { 2571 MUTEX_LOCK_ENSURE(ctx, q_mutex); 2572 if (edge->stat == EDGE_IDLE) { 2573 grn_com_queue_enque(ctx, &ctx_new, (grn_com_queue_entry *)edge); 2574 edge->stat = EDGE_WAIT; 2575 if (n_floating_threads == 0 && n_running_threads < max_n_floating_threads) { 2576 grn_thread thread; 2577 n_running_threads++; 2578 if (THREAD_CREATE(thread, g_worker, NULL)) { 2579 n_running_threads--; 2580 SERR("pthread_create"); 2581 } 2582 } 2583 COND_SIGNAL(q_cond); 2584 } 2585 MUTEX_UNLOCK(q_mutex); 2586 } 2587 2588 static void 2589 g_output(grn_ctx *ctx, int flags, void *arg) 2590 { 2591 grn_edge *edge = arg; 2592 grn_com *com = edge->com; 2593 grn_msg *req = edge->msg, *msg = (grn_msg *)ctx->impl->output.buf; 2594 msg->edge_id = req->edge_id; 2595 msg->header.proto = req->header.proto == GRN_COM_PROTO_MBREQ 2596 ? GRN_COM_PROTO_MBRES : req->header.proto; 2597 if (ctx->rc != GRN_SUCCESS && GRN_BULK_VSIZE(ctx->impl->output.buf) == 0) { 2598 GRN_TEXT_PUTS(ctx, ctx->impl->output.buf, ctx->errbuf); 2599 } 2600 if (grn_msg_send(ctx, (grn_obj *)msg, 2601 (flags & GRN_CTX_MORE) ? GRN_CTX_MORE : GRN_CTX_TAIL)) { 2602 edge->stat = EDGE_ABORT; 2603 } 2604 ctx->impl->output.buf = grn_msg_open(ctx, com, &edge->send_old); 2605 } 2606 2607 static void 2608 g_handler(grn_ctx *ctx, grn_obj *msg) 2609 { 2610 grn_edge *edge; 2611 grn_com *com = ((grn_msg *)msg)->u.peer; 2612 if (ctx->rc) { 2613 if (com->has_sid) { 2614 if ((edge = com->opaque)) { 2615 MUTEX_LOCK_ENSURE(ctx, q_mutex); 2616 if (edge->stat == EDGE_IDLE) { 2617 grn_com_queue_enque(ctx, &ctx_old, (grn_com_queue_entry *)edge); 2618 } 2619 edge->stat = EDGE_ABORT; 2620 MUTEX_UNLOCK(q_mutex); 2621 } else { 2622 grn_com_close(ctx, com); 2623 } 2624 } 2625 grn_msg_close(ctx, msg); 2626 } else { 2627 int added; 2628 edge = grn_edges_add(ctx, &((grn_msg *)msg)->edge_id, &added); 2629 if (added) { 2630 grn_ctx_init(&edge->ctx, 0); 2631 GRN_COM_QUEUE_INIT(&edge->recv_new); 2632 GRN_COM_QUEUE_INIT(&edge->send_old); 2633 grn_ctx_use(&edge->ctx, (grn_obj *)com->ev->opaque); 2634 grn_ctx_recv_handler_set(&edge->ctx, g_output, edge); 2635 com->opaque = edge; 2636 grn_obj_close(&edge->ctx, edge->ctx.impl->output.buf); 2637 edge->ctx.impl->output.buf = 2638 grn_msg_open(&edge->ctx, com, &edge->send_old); 2639 edge->com = com; 2640 edge->stat = EDGE_IDLE; 2641 edge->flags = GRN_EDGE_WORKER; 2642 } 2643 if (edge->ctx.stat == GRN_CTX_QUIT || edge->stat == EDGE_ABORT) { 2644 grn_msg_close(ctx, msg); 2645 } else { 2646 grn_com_queue_enque(ctx, &edge->recv_new, (grn_com_queue_entry *)msg); 2647 g_dispatcher(ctx, edge); 2648 } 2649 } 2650 } 2651 2652 static int 2653 g_server(char *path) 2654 { 2655 int exit_code = EXIT_FAILURE; 2656 grn_ctx ctx_, *ctx = &ctx_; 2657 grn_ctx_init(ctx, 0); 2658 GRN_COM_QUEUE_INIT(&ctx_new); 2659 GRN_COM_QUEUE_INIT(&ctx_old); 2660 check_rlimit_nofile(ctx); 2661 exit_code = start_service(ctx, path, g_dispatcher, g_handler); 2662 grn_ctx_fin(ctx); 2663 return exit_code; 2664 } 2665 2666 enum { 2667 ACTION_USAGE = 1, 2668 ACTION_VERSION, 2669 ACTION_SHOW_CONFIG, 2670 ACTION_ERROR 2671 }; 2672 2673 #define ACTION_MASK (0x0f) 2674 #define MODE_MASK (0xf0) 2675 #define FLAG_MODE_ALONE (1 << 4) 2676 #define FLAG_MODE_CLIENT (1 << 5) 2677 #define FLAG_MODE_DAEMON (1 << 6) 2678 #define FLAG_MODE_SERVER (1 << 7) 2679 #define FLAG_NEW_DB (1 << 8) 2680 #define FLAG_USE_WINDOWS_EVENT_LOG (1 << 9) 2681 2682 static uint32_t 2683 get_core_number(void) 2684 { 2685 #ifdef WIN32 2686 SYSTEM_INFO sinfo; 2687 GetSystemInfo(&sinfo); 2688 return sinfo.dwNumberOfProcessors; 2689 #else /* WIN32 */ 2690 # ifdef _SC_NPROCESSORS_CONF 2691 return sysconf(_SC_NPROCESSORS_CONF); 2692 # else 2693 int n_processors; 2694 size_t length = sizeof(n_processors); 2695 int mib[] = {CTL_HW, HW_NCPU}; 2696 if (sysctl(mib, sizeof(mib) / sizeof(mib[0]), 2697 &n_processors, &length, NULL, 0) == 0 && 2698 length == sizeof(n_processors) && 2699 0 < n_processors) { 2700 return n_processors; 2701 } else { 2702 return 1; 2703 } 2704 # endif /* _SC_NPROCESSORS_CONF */ 2705 #endif /* WIN32 */ 2706 } 2707 2708 /* 2709 * The length of each line, including an end-of-line, in config file should be 2710 * shorter than (CONFIG_FILE_BUF_SIZE - 1) bytes. Too long lines are ignored. 2711 * Note that both '\r' and '\n' are handled as end-of-lines. 2712 * 2713 * '#' and ';' are special symbols to start comments. A comment ends with an 2714 * end-of-line. 2715 * 2716 * Format: name[=value] 2717 * - Preceding/trailing white-spaces of each line are removed. 2718 * - White-spaces aroung '=' are removed. 2719 * - name does not allow white-spaces. 2720 */ 2721 #define CONFIG_FILE_BUF_SIZE 4096 2722 #define CONFIG_FILE_MAX_NAME_LENGTH 128 2723 #define CONFIG_FILE_MAX_VALUE_LENGTH 2048 2724 2725 typedef enum { 2726 CONFIG_FILE_SUCCESS, 2727 CONFIG_FILE_FORMAT_ERROR, 2728 CONFIG_FILE_FOPEN_ERROR, 2729 CONFIG_FILE_MALLOC_ERROR, 2730 CONFIG_FILE_ATEXIT_ERROR 2731 } config_file_status; 2732 2733 /* 2734 * The node type of a linked list for storing values. Note that a value is 2735 * stored in the extra space of an object. 2736 */ 2737 typedef struct _config_file_entry { 2738 struct _config_file_entry *next; 2739 } config_file_entry; 2740 2741 static config_file_entry *config_file_entry_head = NULL; 2742 2743 static void 2744 config_file_clear(void) { 2745 while (config_file_entry_head) { 2746 config_file_entry *next = config_file_entry_head->next; 2747 free(config_file_entry_head); 2748 config_file_entry_head = next; 2749 } 2750 } 2751 2752 static config_file_status 2753 config_file_register(const char *path, const grn_str_getopt_opt *opts, 2754 int *flags, const char *name, size_t name_length, 2755 const char *value, size_t value_length) 2756 { 2757 char name_buf[CONFIG_FILE_MAX_NAME_LENGTH + 3]; 2758 config_file_entry *entry = NULL; 2759 char *args[4]; 2760 2761 name_buf[0] = name_buf[1] = '-'; 2762 grn_strcpy(name_buf + 2, CONFIG_FILE_MAX_NAME_LENGTH + 1, name); 2763 2764 if (value) { 2765 const size_t entry_size = sizeof(config_file_entry) + value_length + 1; 2766 entry = (config_file_entry *)malloc(entry_size); 2767 if (!entry) { 2768 fprintf(stderr, "memory allocation failed: %u bytes\n", 2769 (unsigned int)entry_size); 2770 return CONFIG_FILE_MALLOC_ERROR; 2771 } 2772 grn_strcpy((char *)(entry + 1), value_length + 1, value); 2773 entry->next = config_file_entry_head; 2774 if (!config_file_entry_head) { 2775 if (atexit(config_file_clear)) { 2776 free(entry); 2777 return CONFIG_FILE_ATEXIT_ERROR; 2778 } 2779 } 2780 config_file_entry_head = entry; 2781 } 2782 2783 args[0] = (char *)path; 2784 args[1] = name_buf; 2785 args[2] = entry ? (char *)(entry + 1) : NULL; 2786 args[3] = NULL; 2787 grn_str_getopt(entry ? 3 : 2, args, opts, flags); 2788 return CONFIG_FILE_SUCCESS; 2789 } 2790 2791 static config_file_status 2792 config_file_parse(const char *path, const grn_str_getopt_opt *opts, 2793 int *flags, char *buf) { 2794 char *ptr, *name, *value; 2795 size_t name_length, value_length; 2796 2797 while (isspace((unsigned char)*buf)) { 2798 buf++; 2799 } 2800 2801 ptr = buf; 2802 while (*ptr && *ptr != '#' && *ptr != ';') { 2803 ptr++; 2804 } 2805 2806 do { 2807 *ptr-- = '\0'; 2808 } while (ptr >= buf && isspace((unsigned char)*ptr)); 2809 2810 if (!*buf) { 2811 return CONFIG_FILE_SUCCESS; 2812 } 2813 2814 name = ptr = buf; 2815 while (*ptr && !isspace((unsigned char)*ptr) && *ptr != '=') { 2816 ptr++; 2817 } 2818 while (isspace((unsigned char)*ptr)) { 2819 *ptr++ = '\0'; 2820 } 2821 2822 name_length = strlen(name); 2823 if (name_length == 0) { 2824 return CONFIG_FILE_SUCCESS; 2825 } else if (name_length > CONFIG_FILE_MAX_NAME_LENGTH) { 2826 fprintf(stderr, "too long name in config file: %u bytes\n", 2827 (unsigned int)name_length); 2828 return CONFIG_FILE_FORMAT_ERROR; 2829 } 2830 2831 if (*ptr == '=') { 2832 *ptr++ = '\0'; 2833 while (isspace((unsigned char)*ptr)) { 2834 ptr++; 2835 } 2836 value = ptr; 2837 } else if (*ptr) { 2838 fprintf(stderr, "invalid name in config file\n"); 2839 return CONFIG_FILE_FORMAT_ERROR; 2840 } else { 2841 value = NULL; 2842 } 2843 2844 value_length = value ? strlen(value) : 0; 2845 if (value_length > CONFIG_FILE_MAX_VALUE_LENGTH) { 2846 fprintf(stderr, "too long value in config file: %u bytes\n", 2847 (unsigned int)value_length); 2848 return CONFIG_FILE_FORMAT_ERROR; 2849 } 2850 2851 return config_file_register(path, opts, flags, 2852 name, name_length, value, value_length); 2853 } 2854 2855 static config_file_status 2856 config_file_load(const char *path, const grn_str_getopt_opt *opts, int *flags) 2857 { 2858 config_file_status status = CONFIG_FILE_SUCCESS; 2859 char buf[CONFIG_FILE_BUF_SIZE]; 2860 size_t length = 0; 2861 FILE * const file = fopen(path, "rb"); 2862 if (!file) { 2863 return CONFIG_FILE_FOPEN_ERROR; 2864 } 2865 2866 for ( ; ; ) { 2867 int c = fgetc(file); 2868 if (c == '\r' || c == '\n' || c == EOF) { 2869 if (length < sizeof(buf) - 1) { 2870 buf[length] = '\0'; 2871 status = config_file_parse(path, opts, flags, buf); 2872 if (status != CONFIG_FILE_SUCCESS) { 2873 break; 2874 } 2875 } 2876 length = 0; 2877 } else if (c == '\0') { 2878 fprintf(stderr, "prohibited '\\0' in config file: %s\n", path); 2879 status = CONFIG_FILE_FORMAT_ERROR; 2880 break; 2881 } else { 2882 if (length < sizeof(buf) - 1) { 2883 buf[length] = (char)c; 2884 } 2885 length++; 2886 } 2887 2888 if (c == EOF) { 2889 break; 2890 } 2891 } 2892 2893 fclose(file); 2894 return status; 2895 } 2896 2897 static const int default_http_port = DEFAULT_HTTP_PORT; 2898 static const int default_gqtp_port = DEFAULT_GQTP_PORT; 2899 static grn_encoding default_encoding = GRN_ENC_DEFAULT; 2900 static uint32_t default_max_n_threads = DEFAULT_MAX_N_FLOATING_THREADS; 2901 static const grn_log_level default_log_level = GRN_LOG_DEFAULT_LEVEL; 2902 static const char * const default_protocol = "gqtp"; 2903 static const char *default_hostname = "localhost"; 2904 static const char * const default_dest = "localhost"; 2905 static const char *default_log_path = ""; 2906 static const char *default_query_log_path = ""; 2907 static const char *default_config_path = ""; 2908 static const char *default_document_root = ""; 2909 static grn_command_version default_default_command_version = 2910 GRN_COMMAND_VERSION_DEFAULT; 2911 static int64_t default_default_match_escalation_threshold = 0; 2912 static const char * const default_bind_address = "0.0.0.0"; 2913 static double default_default_request_timeout = 0.0; 2914 2915 static void 2916 init_default_hostname(void) 2917 { 2918 static char hostname[HOST_NAME_MAX + 1]; 2919 struct addrinfo hints, *result; 2920 2921 hostname[HOST_NAME_MAX] = '\0'; 2922 if (gethostname(hostname, HOST_NAME_MAX) == -1) 2923 return; 2924 2925 memset(&hints, 0, sizeof(hints)); 2926 hints.ai_family = AF_UNSPEC; 2927 hints.ai_socktype = SOCK_STREAM; 2928 hints.ai_addr = NULL; 2929 hints.ai_canonname = NULL; 2930 hints.ai_next = NULL; 2931 if (getaddrinfo(hostname, NULL, &hints, &result) != 0) 2932 return; 2933 freeaddrinfo(result); 2934 2935 default_hostname = hostname; 2936 } 2937 2938 static void 2939 init_default_settings(void) 2940 { 2941 output = stdout; 2942 2943 default_encoding = grn_encoding_parse(GRN_DEFAULT_ENCODING); 2944 2945 { 2946 const uint32_t n_cores = get_core_number(); 2947 if (n_cores != 0) { 2948 default_max_n_threads = n_cores; 2949 } 2950 } 2951 2952 init_default_hostname(); 2953 2954 default_log_path = grn_default_logger_get_path(); 2955 default_query_log_path = grn_default_query_logger_get_path(); 2956 2957 default_config_path = getenv("GRN_CONFIG_PATH"); 2958 if (!default_config_path) { 2959 default_config_path = GRN_CONFIG_PATH; 2960 if (!default_config_path) { 2961 default_config_path = ""; 2962 } 2963 } 2964 2965 #ifdef WIN32 2966 { 2967 static char windows_default_document_root[PATH_MAX]; 2968 size_t document_root_length = strlen(grn_windows_base_dir()) + 1 + 2969 strlen(GRN_DEFAULT_RELATIVE_DOCUMENT_ROOT) + 1; 2970 if (document_root_length >= PATH_MAX) { 2971 fprintf(stderr, "can't use default root: too long path\n"); 2972 } else { 2973 grn_strcpy(windows_default_document_root, PATH_MAX, 2974 grn_windows_base_dir()); 2975 grn_strcat(windows_default_document_root, PATH_MAX, 2976 "/"); 2977 grn_strcat(windows_default_document_root, PATH_MAX, 2978 GRN_DEFAULT_RELATIVE_DOCUMENT_ROOT); 2979 default_document_root = windows_default_document_root; 2980 } 2981 } 2982 #else 2983 default_document_root = GRN_DEFAULT_DOCUMENT_ROOT; 2984 #endif 2985 2986 default_default_command_version = grn_get_default_command_version(); 2987 default_default_match_escalation_threshold = 2988 grn_get_default_match_escalation_threshold(); 2989 default_default_request_timeout = grn_get_default_request_timeout(); 2990 } 2991 2992 static void 2993 show_config(FILE *out, const grn_str_getopt_opt *opts, int flags) 2994 { 2995 const grn_str_getopt_opt *o; 2996 2997 for (o = opts; o->opt || o->longopt; o++) { 2998 switch (o->op) { 2999 case GETOPT_OP_NONE: 3000 if (o->arg && *o->arg) { 3001 if (o->longopt && strcmp(o->longopt, "config-path")) { 3002 fprintf(out, "%s=%s\n", o->longopt, *o->arg); 3003 } 3004 } 3005 break; 3006 case GETOPT_OP_ON: 3007 if (flags & o->flag) { 3008 goto no_arg; 3009 } 3010 break; 3011 case GETOPT_OP_OFF: 3012 if (!(flags & o->flag)) { 3013 goto no_arg; 3014 } 3015 break; 3016 case GETOPT_OP_UPDATE: 3017 if (flags == o->flag) { 3018 no_arg: 3019 if (o->longopt) { 3020 fprintf(out, "%s\n", o->longopt); 3021 } 3022 } 3023 break; 3024 } 3025 } 3026 } 3027 3028 static void 3029 show_version(void) 3030 { 3031 printf("%s %s [", 3032 grn_get_package_label(), 3033 grn_get_version()); 3034 3035 /* FIXME: Should we detect host information dynamically on Windows? */ 3036 #ifdef HOST_OS 3037 printf("%s,", HOST_OS); 3038 #endif 3039 #ifdef HOST_CPU 3040 printf("%s,", HOST_CPU); 3041 #endif 3042 printf("%s", GRN_DEFAULT_ENCODING); 3043 3044 printf(",match-escalation-threshold=%" GRN_FMT_LLD, 3045 grn_get_default_match_escalation_threshold()); 3046 3047 #ifndef NO_NFKC 3048 printf(",nfkc"); 3049 #endif 3050 #ifdef GRN_WITH_MECAB 3051 printf(",mecab"); 3052 #endif 3053 #ifdef GRN_WITH_MESSAGE_PACK 3054 printf(",msgpack"); 3055 #endif 3056 #ifdef GRN_WITH_MRUBY 3057 printf(",mruby"); 3058 #endif 3059 #ifdef GRN_WITH_ONIGMO 3060 printf(",onigmo"); 3061 #endif 3062 #ifdef GRN_WITH_ZLIB 3063 printf(",zlib"); 3064 #endif 3065 #ifdef GRN_WITH_LZ4 3066 printf(",lz4"); 3067 #endif 3068 #ifdef GRN_WITH_ZSTD 3069 printf(",zstd"); 3070 #endif 3071 #ifdef USE_KQUEUE 3072 printf(",kqueue"); 3073 #endif 3074 #ifdef USE_EPOLL 3075 printf(",epoll"); 3076 #endif 3077 #ifdef USE_POLL 3078 printf(",poll"); 3079 #endif 3080 printf("]\n"); 3081 3082 #ifdef CONFIGURE_OPTIONS 3083 printf("\n"); 3084 printf("configure options: <%s>\n", CONFIGURE_OPTIONS); 3085 #endif 3086 } 3087 3088 static void 3089 show_usage(FILE *output) 3090 { 3091 uint32_t default_cache_limit = GRN_CACHE_DEFAULT_MAX_N_ENTRIES; 3092 3093 fprintf(output, 3094 "Usage: groonga [options...] [dest]\n" 3095 "\n" 3096 "Mode options: (default: standalone)\n" 3097 " By default, groonga runs in standalone mode.\n" 3098 " -c: run in client mode\n" 3099 " -s: run in server mode\n" 3100 " -d: run in daemon mode\n" 3101 "\n" 3102 "Database creation options:\n" 3103 " -n: create new database (except client mode)\n" 3104 " -e, --encoding <encoding>:\n" 3105 " specify encoding for new database\n" 3106 " [none|euc|utf8|sjis|latin1|koi8r] (default: %s)\n" 3107 "\n" 3108 "Standalone/client options:\n" 3109 " --file <path>: read commands from specified file\n" 3110 " --input-fd <FD>: read commands from specified file descriptor\n" 3111 " --file has a prioriry over --input-fd\n" 3112 " --output-fd <FD>: output response to specified file descriptor\n" 3113 " -p, --port <port number>: specify server port number (client mode only)\n" 3114 " (default: %d)\n" 3115 "\n" 3116 "Server/daemon options:\n" 3117 " --bind-address <ip/hostname>:\n" 3118 " specify server address to bind\n" 3119 " (default: %s)\n" 3120 " -p, --port <port number>: specify server port number\n" 3121 " (HTTP default: %d, GQTP default: %d)\n" 3122 " -i, --server-id <ip/hostname>:\n" 3123 " specify server ID address (default: %s)\n" 3124 " --protocol <protocol>: specify server protocol to listen\n" 3125 " [gqtp|http|memcached] (default: %s)\n" 3126 " --document-root <path>: specify document root path (http only)\n" 3127 " (default: %s)\n" 3128 " --cache-limit <limit>: specify max number of cache data (default: %u)\n" 3129 " -t, --max-threads <max threads>:\n" 3130 " specify max number of threads (default: %u)\n" 3131 " --pid-path <path>: specify file to write process ID to\n" 3132 " (daemon mode only)\n" 3133 " --default-request-timeout <timeout>:\n" 3134 " specify the default request timeout in seconds\n" 3135 " (default: %f)\n" 3136 " --cache-base-path <path>: specify the cache base path\n" 3137 " You can make cache persistent by this option\n" 3138 " You must specify path on memory file system\n" 3139 " (default: none; disabled)\n" 3140 "\n" 3141 "Memcached options:\n" 3142 " --memcached-column <column>:\n" 3143 " specify column to access by memcached protocol\n" 3144 " The column must be text type column and\n" 3145 " its table must be not NO_KEY table\n" 3146 "\n" 3147 "Logging options:\n" 3148 " -l, --log-level <log level>:\n" 3149 " specify log level\n" 3150 " [none|emergency|alert|critical|\n" 3151 " error|warning|notice|info|debug|dump]\n" 3152 " (default: %s)\n" 3153 " --log-path <path>: specify log path\n" 3154 " (default: %s)\n" 3155 " --log-rotate-threshold-size <threshold>:\n" 3156 " specify threshold for log rotate\n" 3157 " Log file is rotated when\n" 3158 " log file size is larger than or\n" 3159 " equals to the threshold\n" 3160 " (default: 0; disabled)\n" 3161 #ifdef WIN32 3162 " --use-windows-event-log:\n" 3163 " report logs as Windows events\n" 3164 #endif /* WIN32 */ 3165 " --query-log-path <path>:\n" 3166 " specify query log path\n" 3167 " (default: %s)\n" 3168 " --query-log-rotate-threshold-size <threshold>:\n" 3169 " specify threshold for query log rotate\n" 3170 " Query log file is rotated when\n" 3171 " query log file size is larger than or\n" 3172 " equals to the threshold\n" 3173 " (default: 0; disabled)\n" 3174 "\n" 3175 "Common options:\n" 3176 " --working-directory <path>:\n" 3177 " specify working directory path\n" 3178 " (none)\n" 3179 " --config-path <path>:\n" 3180 " specify config file path\n" 3181 " (default: %s)\n" 3182 " --default-command-version <version>:\n" 3183 " specify default command version (default: %d)\n" 3184 " --default-match-escalation-threshold <threshold>:\n" 3185 " specify default match escalation threshold" 3186 " (default: %" GRN_FMT_LLD ")\n" 3187 "\n" 3188 " --show-config: show config\n" 3189 " -h, --help: show usage\n" 3190 " --version: show groonga version\n" 3191 "\n" 3192 "dest:\n" 3193 " <db pathname> [<commands>]: in standalone mode\n" 3194 " <db pathname>: in server/daemon mode\n" 3195 " <dest hostname> [<commands>]: in client mode (default: %s)\n", 3196 grn_encoding_to_string(default_encoding), 3197 default_gqtp_port, default_bind_address, 3198 default_http_port, default_gqtp_port, default_hostname, default_protocol, 3199 default_document_root, default_cache_limit, default_max_n_threads, 3200 default_default_request_timeout, 3201 grn_log_level_to_string(default_log_level), 3202 default_log_path, default_query_log_path, 3203 default_config_path, default_default_command_version, 3204 (long long int)default_default_match_escalation_threshold, 3205 default_dest); 3206 } 3207 3208 int 3209 main(int argc, char **argv) 3210 { 3211 const char *port_arg = NULL; 3212 const char *encoding_arg = NULL; 3213 const char *max_n_threads_arg = NULL; 3214 const char *log_level_arg = NULL; 3215 const char *bind_address_arg = NULL; 3216 const char *hostname_arg = NULL; 3217 const char *protocol_arg = NULL; 3218 const char *log_path_arg = GRN_LOG_PATH; 3219 const char *log_rotate_threshold_size_arg = NULL; 3220 const char *query_log_path_arg = NULL; 3221 const char *query_log_rotate_threshold_size_arg = NULL; 3222 const char *cache_limit_arg = NULL; 3223 const char *document_root_arg = NULL; 3224 const char *default_command_version_arg = NULL; 3225 const char *default_match_escalation_threshold_arg = NULL; 3226 const char *input_fd_arg = NULL; 3227 const char *output_fd_arg = NULL; 3228 const char *working_directory_arg = NULL; 3229 const char *config_path = NULL; 3230 const char *default_request_timeout_arg = NULL; 3231 const char *cache_base_path = NULL; 3232 int exit_code = EXIT_SUCCESS; 3233 int i; 3234 int flags = 0; 3235 uint32_t cache_limit = 0; 3236 grn_command_version default_command_version; 3237 int64_t default_match_escalation_threshold = 0; 3238 double default_request_timeout = 0.0; 3239 grn_bool need_line_editor = GRN_FALSE; 3240 static grn_str_getopt_opt opts[] = { 3241 {'p', "port", NULL, 0, GETOPT_OP_NONE}, 3242 {'e', "encoding", NULL, 0, GETOPT_OP_NONE}, 3243 {'t', "max-threads", NULL, 0, GETOPT_OP_NONE}, 3244 {'h', "help", NULL, ACTION_USAGE, GETOPT_OP_UPDATE}, 3245 {'c', NULL, NULL, FLAG_MODE_CLIENT, GETOPT_OP_ON}, 3246 {'d', NULL, NULL, FLAG_MODE_DAEMON, GETOPT_OP_ON}, 3247 {'s', NULL, NULL, FLAG_MODE_SERVER, GETOPT_OP_ON}, 3248 {'l', "log-level", NULL, 0, GETOPT_OP_NONE}, 3249 {'i', "server-id", NULL, 0, GETOPT_OP_NONE}, 3250 {'n', NULL, NULL, FLAG_NEW_DB, GETOPT_OP_ON}, 3251 {'\0', "protocol", NULL, 0, GETOPT_OP_NONE}, 3252 {'\0', "version", NULL, ACTION_VERSION, GETOPT_OP_UPDATE}, 3253 {'\0', "log-path", NULL, 0, GETOPT_OP_NONE}, 3254 {'\0', "log-rotate-threshold-size", NULL, 0, GETOPT_OP_NONE}, 3255 {'\0', "query-log-path", NULL, 0, GETOPT_OP_NONE}, 3256 {'\0', "query-log-rotate-threshold-size", NULL, 0, GETOPT_OP_NONE}, 3257 {'\0', "pid-path", NULL, 0, GETOPT_OP_NONE}, 3258 {'\0', "config-path", NULL, 0, GETOPT_OP_NONE}, 3259 {'\0', "show-config", NULL, ACTION_SHOW_CONFIG, GETOPT_OP_UPDATE}, 3260 {'\0', "cache-limit", NULL, 0, GETOPT_OP_NONE}, 3261 {'\0', "file", NULL, 0, GETOPT_OP_NONE}, 3262 {'\0', "document-root", NULL, 0, GETOPT_OP_NONE}, 3263 {'\0', "default-command-version", NULL, 0, GETOPT_OP_NONE}, 3264 {'\0', "default-match-escalation-threshold", NULL, 0, GETOPT_OP_NONE}, 3265 {'\0', "bind-address", NULL, 0, GETOPT_OP_NONE}, 3266 {'\0', "input-fd", NULL, 0, GETOPT_OP_NONE}, 3267 {'\0', "output-fd", NULL, 0, GETOPT_OP_NONE}, 3268 {'\0', "working-directory", NULL, 0, GETOPT_OP_NONE}, 3269 {'\0', "use-windows-event-log", NULL, 3270 FLAG_USE_WINDOWS_EVENT_LOG, GETOPT_OP_ON}, 3271 {'\0', "memcached-column", NULL, 0, GETOPT_OP_NONE}, 3272 {'\0', "default-request-timeout", NULL, 0, GETOPT_OP_NONE}, 3273 {'\0', "cache-base-path", NULL, 0, GETOPT_OP_NONE}, 3274 {'\0', NULL, NULL, 0, 0} 3275 }; 3276 opts[0].arg = &port_arg; 3277 opts[1].arg = &encoding_arg; 3278 opts[2].arg = &max_n_threads_arg; 3279 opts[7].arg = &log_level_arg; 3280 opts[8].arg = &hostname_arg; 3281 opts[10].arg = &protocol_arg; 3282 opts[12].arg = &log_path_arg; 3283 opts[13].arg = &log_rotate_threshold_size_arg; 3284 opts[14].arg = &query_log_path_arg; 3285 opts[15].arg = &query_log_rotate_threshold_size_arg; 3286 opts[16].arg = &pid_file_path; 3287 opts[17].arg = &config_path; 3288 opts[19].arg = &cache_limit_arg; 3289 opts[20].arg = &input_path; 3290 opts[21].arg = &document_root_arg; 3291 opts[22].arg = &default_command_version_arg; 3292 opts[23].arg = &default_match_escalation_threshold_arg; 3293 opts[24].arg = &bind_address_arg; 3294 opts[25].arg = &input_fd_arg; 3295 opts[26].arg = &output_fd_arg; 3296 opts[27].arg = &working_directory_arg; 3297 opts[29].arg = &memcached_column_name; 3298 opts[30].arg = &default_request_timeout_arg; 3299 opts[31].arg = &cache_base_path; 3300 3301 reset_ready_notify_pipe(); 3302 3303 init_default_settings(); 3304 3305 /* only for parsing --config-path. */ 3306 i = grn_str_getopt(argc, argv, opts, &flags); 3307 if (i < 0) { 3308 show_usage(stderr); 3309 return EXIT_FAILURE; 3310 } 3311 3312 if (config_path) { 3313 const config_file_status status = config_file_load(config_path, opts, &flags); 3314 if (status == CONFIG_FILE_FOPEN_ERROR) { 3315 fprintf(stderr, "%s: can't open config file: %s (%s)\n", 3316 argv[0], config_path, strerror(errno)); 3317 return EXIT_FAILURE; 3318 } else if (status != CONFIG_FILE_SUCCESS) { 3319 fprintf(stderr, "%s: failed to parse config file: %s (%s)\n", 3320 argv[0], config_path, 3321 (status == CONFIG_FILE_FORMAT_ERROR) ? "Invalid format" : strerror(errno)); 3322 return EXIT_FAILURE; 3323 } 3324 } else if (*default_config_path) { 3325 const config_file_status status = 3326 config_file_load(default_config_path, opts, &flags); 3327 if (status != CONFIG_FILE_SUCCESS && status != CONFIG_FILE_FOPEN_ERROR) { 3328 fprintf(stderr, "%s: failed to parse config file: %s (%s)\n", 3329 argv[0], default_config_path, 3330 (status == CONFIG_FILE_FORMAT_ERROR) ? "Invalid format" : strerror(errno)); 3331 return EXIT_FAILURE; 3332 } 3333 } 3334 3335 if (working_directory_arg) { 3336 if (chdir(working_directory_arg) == -1) { 3337 fprintf(stderr, "%s: failed to change directory: %s: %s\n", 3338 argv[0], working_directory_arg, strerror(errno)); 3339 return EXIT_FAILURE; 3340 } 3341 } 3342 3343 if (cache_base_path) { 3344 grn_set_default_cache_base_path(cache_base_path); 3345 } 3346 3347 /* ignore mode option in config file */ 3348 flags = (flags == ACTION_ERROR) ? 0 : (flags & ~ACTION_MASK); 3349 3350 i = grn_str_getopt(argc, argv, opts, &flags); 3351 if (i < 0) { flags = ACTION_ERROR; } 3352 switch (flags & ACTION_MASK) { 3353 case ACTION_VERSION : 3354 show_version(); 3355 return EXIT_SUCCESS; 3356 case ACTION_USAGE : 3357 show_usage(output); 3358 return EXIT_SUCCESS; 3359 case ACTION_SHOW_CONFIG : 3360 show_config(output, opts, flags & ~ACTION_MASK); 3361 return EXIT_SUCCESS; 3362 case ACTION_ERROR : 3363 show_usage(stderr); 3364 return EXIT_FAILURE; 3365 } 3366 3367 if ((flags & MODE_MASK) == 0) { 3368 flags |= FLAG_MODE_ALONE; 3369 } 3370 3371 if (port_arg) { 3372 const char * const end = port_arg + strlen(port_arg); 3373 const char *rest = NULL; 3374 const int value = grn_atoi(port_arg, end, &rest); 3375 if (rest != end || value <= 0 || value > 65535) { 3376 fprintf(stderr, "invalid port number: <%s>\n", port_arg); 3377 return EXIT_FAILURE; 3378 } 3379 port = value; 3380 } else { 3381 if (protocol_arg) { 3382 if (*protocol_arg == 'h' || *protocol_arg == 'H') { 3383 port = default_http_port; 3384 } 3385 } 3386 } 3387 3388 if (encoding_arg) { 3389 switch (*encoding_arg) { 3390 case 'n' : 3391 case 'N' : 3392 encoding = GRN_ENC_NONE; 3393 break; 3394 case 'e' : 3395 case 'E' : 3396 encoding = GRN_ENC_EUC_JP; 3397 break; 3398 case 'u' : 3399 case 'U' : 3400 encoding = GRN_ENC_UTF8; 3401 break; 3402 case 's' : 3403 case 'S' : 3404 encoding = GRN_ENC_SJIS; 3405 break; 3406 case 'l' : 3407 case 'L' : 3408 encoding = GRN_ENC_LATIN1; 3409 break; 3410 case 'k' : 3411 case 'K' : 3412 encoding = GRN_ENC_KOI8R; 3413 break; 3414 default: 3415 encoding = GRN_ENC_DEFAULT; 3416 break; 3417 } 3418 } else { 3419 encoding = GRN_ENC_DEFAULT; 3420 } 3421 3422 if (!grn_document_root) { 3423 grn_document_root = default_document_root; 3424 } 3425 3426 if (protocol_arg) { 3427 switch (*protocol_arg) { 3428 case 'g' : 3429 case 'G' : 3430 do_client = g_client; 3431 do_server = g_server; 3432 break; 3433 case 'h' : 3434 case 'H' : 3435 do_client = g_client; 3436 do_server = h_server; 3437 break; 3438 case 'm' : 3439 case 'M' : 3440 is_memcached_mode = GRN_TRUE; 3441 do_client = g_client; 3442 do_server = g_server; 3443 break; 3444 default : 3445 do_client = g_client; 3446 do_server = g_server; 3447 break; 3448 } 3449 } else { 3450 do_client = g_client; 3451 do_server = g_server; 3452 } 3453 3454 #ifdef WIN32 3455 if (flags & FLAG_USE_WINDOWS_EVENT_LOG) { 3456 use_windows_event_log = GRN_TRUE; 3457 } 3458 #endif /* WIN32 */ 3459 3460 if (use_windows_event_log) { 3461 grn_windows_event_logger_set(NULL, windows_event_source_name); 3462 } 3463 3464 if (log_path_arg) { 3465 grn_default_logger_set_path(log_path_arg); 3466 } 3467 3468 if (log_rotate_threshold_size_arg) { 3469 const char * const end = 3470 log_rotate_threshold_size_arg + 3471 strlen(log_rotate_threshold_size_arg); 3472 const char *rest = NULL; 3473 const uint64_t value = grn_atoull(log_rotate_threshold_size_arg, end, &rest); 3474 if (end != rest) { 3475 fprintf(stderr, "invalid log rotate threshold size: <%s>\n", 3476 log_rotate_threshold_size_arg); 3477 return EXIT_FAILURE; 3478 } 3479 grn_default_logger_set_rotate_threshold_size(value); 3480 } 3481 3482 if (query_log_path_arg) { 3483 grn_default_query_logger_set_path(query_log_path_arg); 3484 } 3485 3486 if (query_log_rotate_threshold_size_arg) { 3487 const char * const end = 3488 query_log_rotate_threshold_size_arg + 3489 strlen(query_log_rotate_threshold_size_arg); 3490 const char *rest = NULL; 3491 const uint64_t value = 3492 grn_atoull(query_log_rotate_threshold_size_arg, end, &rest); 3493 if (end != rest) { 3494 fprintf(stderr, "invalid query log rotate threshold size: <%s>\n", 3495 query_log_rotate_threshold_size_arg); 3496 return EXIT_FAILURE; 3497 } 3498 grn_default_query_logger_set_rotate_threshold_size(value); 3499 } 3500 3501 { 3502 grn_log_level log_level; 3503 3504 if (log_level_arg) { 3505 grn_bool parsed; 3506 3507 parsed = grn_log_level_parse(log_level_arg, &log_level); 3508 if (!parsed) { 3509 const char * const end = log_level_arg + strlen(log_level_arg); 3510 const char *rest = NULL; 3511 const int value = grn_atoi(log_level_arg, end, &rest); 3512 if (end != rest || value < GRN_LOG_NONE || value > GRN_LOG_DUMP) { 3513 fprintf(stderr, "invalid log level: <%s>\n", log_level_arg); 3514 return EXIT_FAILURE; 3515 } 3516 log_level = value; 3517 } 3518 } else { 3519 log_level = default_log_level; 3520 } 3521 3522 grn_default_logger_set_max_level(log_level); 3523 } 3524 3525 if (max_n_threads_arg) { 3526 const char * const end = max_n_threads_arg + strlen(max_n_threads_arg); 3527 const char *rest = NULL; 3528 const uint32_t value = grn_atoui(max_n_threads_arg, end, &rest); 3529 if (end != rest || value < 1 || value > 100) { 3530 fprintf(stderr, "invalid max number of threads: <%s>\n", 3531 max_n_threads_arg); 3532 return EXIT_FAILURE; 3533 } 3534 max_n_floating_threads = value; 3535 } else { 3536 if (flags & FLAG_MODE_ALONE) { 3537 max_n_floating_threads = 1; 3538 } else { 3539 max_n_floating_threads = default_max_n_threads; 3540 } 3541 } 3542 3543 grn_thread_set_get_limit_func(groonga_get_thread_limit, NULL); 3544 grn_thread_set_set_limit_func(groonga_set_thread_limit, NULL); 3545 3546 if (output_fd_arg) { 3547 const char * const end = output_fd_arg + strlen(output_fd_arg); 3548 const char *rest = NULL; 3549 const int output_fd = grn_atoi(output_fd_arg, end, &rest); 3550 if (rest != end || output_fd == 0) { 3551 fprintf(stderr, "invalid output FD: <%s>\n", output_fd_arg); 3552 return EXIT_FAILURE; 3553 } 3554 output = fdopen(output_fd, "w"); 3555 if (!output) { 3556 fprintf(stderr, "can't open output FD: %d (%s)\n", 3557 output_fd, strerror(errno)); 3558 return EXIT_FAILURE; 3559 } 3560 } 3561 3562 3563 if (bind_address_arg) { 3564 const size_t bind_address_length = strlen(bind_address_arg); 3565 if (bind_address_length > HOST_NAME_MAX) { 3566 fprintf(stderr, "too long bind address: %s (%u bytes):" 3567 " must not be longer than %u bytes\n", 3568 bind_address_arg, (unsigned int)bind_address_length, HOST_NAME_MAX); 3569 return EXIT_FAILURE; 3570 } 3571 grn_strcpy(bind_address, HOST_NAME_MAX + 1, bind_address_arg); 3572 } else { 3573 grn_strcpy(bind_address, HOST_NAME_MAX + 1, default_bind_address); 3574 } 3575 3576 if (hostname_arg) { 3577 const size_t hostname_length = strlen(hostname_arg); 3578 if (hostname_length > HOST_NAME_MAX) { 3579 fprintf(stderr, "too long hostname: %s (%u bytes):" 3580 " must not be longer than %u bytes\n", 3581 hostname_arg, (unsigned int)hostname_length, HOST_NAME_MAX); 3582 return EXIT_FAILURE; 3583 } 3584 grn_strcpy(hostname, HOST_NAME_MAX + 1, hostname_arg); 3585 } else { 3586 grn_strcpy(hostname, HOST_NAME_MAX + 1, default_hostname); 3587 } 3588 3589 if (document_root_arg) { 3590 grn_document_root = document_root_arg; 3591 } 3592 3593 if (default_command_version_arg) { 3594 const char * const end = default_command_version_arg 3595 + strlen(default_command_version_arg); 3596 const char *rest = NULL; 3597 const int value = grn_atoi(default_command_version_arg, end, &rest); 3598 if (end != rest || value < GRN_COMMAND_VERSION_MIN || 3599 value > GRN_COMMAND_VERSION_MAX) { 3600 fprintf(stderr, "invalid command version: <%s>\n", 3601 default_command_version_arg); 3602 return EXIT_FAILURE; 3603 } 3604 default_command_version = value; 3605 } else { 3606 default_command_version = default_default_command_version; 3607 } 3608 3609 if (default_match_escalation_threshold_arg) { 3610 const char * const end = default_match_escalation_threshold_arg 3611 + strlen(default_match_escalation_threshold_arg); 3612 const char *rest = NULL; 3613 const int64_t value = grn_atoll(default_match_escalation_threshold_arg, end, &rest); 3614 if (end != rest) { 3615 fprintf(stderr, "invalid match escalation threshold: <%s>\n", 3616 default_match_escalation_threshold_arg); 3617 return EXIT_FAILURE; 3618 } 3619 default_match_escalation_threshold = value; 3620 } else { 3621 default_match_escalation_threshold = default_default_match_escalation_threshold; 3622 } 3623 3624 if (cache_limit_arg) { 3625 const char * const end = cache_limit_arg + strlen(cache_limit_arg); 3626 const char *rest = NULL; 3627 const uint32_t value = grn_atoui(cache_limit_arg, end, &rest); 3628 if (end != rest) { 3629 fprintf(stderr, "invalid --cache-limit value: <%s>\n", cache_limit_arg); 3630 return EXIT_FAILURE; 3631 } 3632 cache_limit = value; 3633 } 3634 3635 if (default_request_timeout_arg) { 3636 const char * const end = 3637 default_request_timeout_arg + strlen(default_request_timeout_arg); 3638 char *rest = NULL; 3639 double value; 3640 value = strtod(default_request_timeout_arg, &rest); 3641 if (end != rest) { 3642 fprintf(stderr, "invalid default request timeout: <%s>\n", 3643 default_request_timeout_arg); 3644 return EXIT_FAILURE; 3645 } 3646 default_request_timeout = value; 3647 } else { 3648 default_request_timeout = default_default_request_timeout; 3649 } 3650 3651 grn_gctx.errbuf[0] = '\0'; 3652 if (grn_init()) { 3653 fprintf(stderr, "failed to initialize Groonga: %s\n", grn_gctx.errbuf); 3654 return EXIT_FAILURE; 3655 } 3656 3657 grn_set_default_encoding(encoding); 3658 3659 if (default_command_version_arg) { 3660 grn_set_default_command_version(default_command_version); 3661 } 3662 3663 if (default_match_escalation_threshold_arg) { 3664 grn_set_default_match_escalation_threshold(default_match_escalation_threshold); 3665 } 3666 3667 if (default_request_timeout_arg) { 3668 grn_set_default_request_timeout(default_request_timeout); 3669 } 3670 3671 grn_set_segv_handler(); 3672 grn_set_int_handler(); 3673 grn_set_term_handler(); 3674 3675 if (cache_limit_arg) { 3676 grn_cache *cache; 3677 cache = grn_cache_current_get(&grn_gctx); 3678 grn_cache_set_max_n_entries(&grn_gctx, cache, cache_limit); 3679 } 3680 3681 MUTEX_INIT(q_mutex); 3682 COND_INIT(q_cond); 3683 3684 if (input_path) { 3685 input_reader = grn_file_reader_open(&grn_gctx, input_path); 3686 if (!input_reader) { 3687 fprintf(stderr, "can't open input file: %s (%s)\n", 3688 input_path, strerror(errno)); 3689 return EXIT_FAILURE; 3690 } 3691 batchmode = GRN_TRUE; 3692 } else { 3693 if (input_fd_arg) { 3694 const char * const end = input_fd_arg + strlen(input_fd_arg); 3695 const char *rest = NULL; 3696 const int input_fd = grn_atoi(input_fd_arg, end, &rest); 3697 if (rest != end || input_fd == 0) { 3698 fprintf(stderr, "invalid input FD: <%s>\n", input_fd_arg); 3699 return EXIT_FAILURE; 3700 } 3701 if (dup2(input_fd, STDIN_FILENO) == -1) { 3702 fprintf(stderr, "can't open input FD: %d (%s)\n", 3703 input_fd, strerror(errno)); 3704 return EXIT_FAILURE; 3705 } 3706 input_reader = grn_file_reader_open(&grn_gctx, "-"); 3707 if (!input_reader) { 3708 fprintf(stderr, "%s", grn_gctx.errbuf); 3709 return EXIT_FAILURE; 3710 } 3711 batchmode = GRN_TRUE; 3712 } else { 3713 input_reader = grn_file_reader_open(&grn_gctx, "-"); 3714 if (!input_reader) { 3715 fprintf(stderr, "%s", grn_gctx.errbuf); 3716 return EXIT_FAILURE; 3717 } 3718 if (argc - i > 1) { 3719 batchmode = GRN_TRUE; 3720 } else { 3721 batchmode = !grn_isatty(0); 3722 } 3723 } 3724 } 3725 3726 if ((flags & (FLAG_MODE_ALONE | FLAG_MODE_CLIENT)) && 3727 !batchmode) { 3728 need_line_editor = GRN_TRUE; 3729 } 3730 3731 #ifdef GRN_WITH_LIBEDIT 3732 if (need_line_editor) { 3733 line_editor_init(argc, argv); 3734 } 3735 #endif 3736 3737 newdb = (flags & FLAG_NEW_DB); 3738 is_daemon_mode = (flags & FLAG_MODE_DAEMON); 3739 if (flags & FLAG_MODE_CLIENT) { 3740 exit_code = do_client(argc - i, argv + i); 3741 } else if (is_daemon_mode || (flags & FLAG_MODE_SERVER)) { 3742 exit_code = do_server(argc > i ? argv[i] : NULL); 3743 } else { 3744 exit_code = do_alone(argc - i, argv + i); 3745 } 3746 3747 COND_FIN(q_cond); 3748 MUTEX_FIN(q_mutex); 3749 3750 if (input_reader) { 3751 grn_file_reader_close(&grn_gctx, input_reader); 3752 } 3753 #ifdef GRN_WITH_LIBEDIT 3754 if (need_line_editor) { 3755 line_editor_fin(); 3756 } 3757 #endif 3758 if (output != stdout) { 3759 fclose(output); 3760 } 3761 grn_fin(); 3762 return exit_code; 3763 } 3764