|
| 1 | +/* |
| 2 | + * Copyright (C) 2020 BiiLabs Co., Ltd. and Contributors |
| 3 | + * All Rights Reserved. |
| 4 | + * This is free software; you can redistribute it and/or modify it under the |
| 5 | + * terms of the MIT license. A copy of the license can be found in the file |
| 6 | + * "LICENSE" at the root of this distribution. |
| 7 | + */ |
| 8 | + |
| 9 | +#include <getopt.h> |
| 10 | +#include "ta_storage.h" |
| 11 | + |
| 12 | +#define logger_id scylladb_logger_id |
| 13 | + |
| 14 | +typedef struct { |
| 15 | + pthread_mutex_t thread_mutex; |
| 16 | + db_permanode_pool_t* pool; |
| 17 | + char* file_path; |
| 18 | +} db_importer_thread_t; |
| 19 | + |
| 20 | +static status_t init_importer_data(db_importer_thread_t* thread_data, db_permanode_pool_t* pool, char* file_list) { |
| 21 | + status_t ret = SC_OK; |
| 22 | + pthread_mutex_init(&thread_data->thread_mutex, NULL); |
| 23 | + thread_data->pool = pool; |
| 24 | + thread_data->file_path = strdup(file_list); |
| 25 | + return ret; |
| 26 | +} |
| 27 | + |
| 28 | +static void* importer_handler(void* data) { |
| 29 | +#define TRANSACTION_BUFFER_SIZE \ |
| 30 | + (NUM_FLEX_TRITS_HASH + 1 + NUM_TRYTES_SERIALIZED_TRANSACTION + 12) // 12 is for snapshot_index |
| 31 | +#define MAX_FILE_PATH 256 |
| 32 | + |
| 33 | + status_t ret = SC_OK; |
| 34 | + db_importer_thread_t* thread_data = (db_importer_thread_t*)data; |
| 35 | + pthread_mutex_lock(&thread_data->thread_mutex); |
| 36 | + FILE* list_file = NULL; |
| 37 | + char file_name_buffer[MAX_FILE_PATH]; |
| 38 | + |
| 39 | + if ((list_file = fopen(thread_data->file_path, "r")) == NULL) { |
| 40 | + /* The specified configuration file does not exist */ |
| 41 | + ret = SC_CONF_FOPEN_ERROR; |
| 42 | + ta_log_error("Failed to open file %s\n", thread_data->file_path); |
| 43 | + goto exit; |
| 44 | + } |
| 45 | + |
| 46 | + while (fgets(file_name_buffer, MAX_FILE_PATH, list_file) != NULL) { |
| 47 | + char input_buffer[TRANSACTION_BUFFER_SIZE]; |
| 48 | + FILE* file = NULL; |
| 49 | + |
| 50 | + int name_len = strlen(file_name_buffer); |
| 51 | + if (name_len > 0) { |
| 52 | + file_name_buffer[name_len - 1] = 0; |
| 53 | + } else { |
| 54 | + ta_log_warning("Empty file name\n"); |
| 55 | + continue; |
| 56 | + } |
| 57 | + |
| 58 | + if ((file = fopen(file_name_buffer, "r")) == NULL) { |
| 59 | + /* The specified configuration file does not exist */ |
| 60 | + ret = SC_CONF_FOPEN_ERROR; |
| 61 | + ta_log_error("Failed to open file %s\n", file_name_buffer); |
| 62 | + goto exit; |
| 63 | + } |
| 64 | + ta_log_info("%s %s\n", "starting to import file : ", file_name_buffer); |
| 65 | + int cnt = 1; |
| 66 | + int cnt_base1000 = 0; |
| 67 | + while (fgets(input_buffer, TRANSACTION_BUFFER_SIZE, file) != NULL) { |
| 68 | + if (cnt % 1000 == 0) { |
| 69 | + ta_log_info("Import %d K transactions\n", ++cnt_base1000); |
| 70 | + cnt = 0; |
| 71 | + } |
| 72 | + if (input_buffer[strlen(input_buffer) - 1] != '\n') { |
| 73 | + ret = SC_STORAGE_INVALID_INPUT; |
| 74 | + ta_log_error("%s\n", "Historical dump file format error"); |
| 75 | + continue; |
| 76 | + } |
| 77 | + |
| 78 | + do { |
| 79 | + ret = db_permanode_thpool_add((tryte_t*)input_buffer, (tryte_t*)input_buffer + NUM_FLEX_TRITS_HASH + 1, |
| 80 | + thread_data->pool); |
| 81 | + if (ret != SC_OK) { |
| 82 | + pthread_cond_wait(&thread_data->pool->finish_request, &thread_data->thread_mutex); |
| 83 | + } |
| 84 | + } while (ret != SC_OK); |
| 85 | + |
| 86 | + cnt++; |
| 87 | + } |
| 88 | + |
| 89 | + ta_log_info("Successfully import file : %s\n", file_name_buffer); |
| 90 | + } |
| 91 | + |
| 92 | +exit: |
| 93 | + if (ret == SC_OK) { |
| 94 | + ta_log_info("%s %s\n", "Successfully import file : ", thread_data->file_path); |
| 95 | + } else { |
| 96 | + ta_log_error("Failed to import file : %s\n", thread_data->file_path); |
| 97 | + } |
| 98 | + return NULL; |
| 99 | +} |
| 100 | + |
| 101 | +int main(int argc, char* argv[]) { |
| 102 | + int thread_num = 1; |
| 103 | + pthread_t* worker_threads; /* thread's structures */ |
| 104 | + pthread_t importer_thread; |
| 105 | + db_worker_thread_t* worker_data; |
| 106 | + db_importer_thread_t importer_data; |
| 107 | + db_permanode_pool_t pool; |
| 108 | + |
| 109 | + char* db_host = "localhost"; |
| 110 | + char* file_path = NULL; |
| 111 | + const struct option longOpt[] = {{"db_host", required_argument, NULL, 's'}, |
| 112 | + {"file", required_argument, NULL, 'f'}, |
| 113 | + {"thread_num", required_argument, NULL, 't'}, |
| 114 | + {NULL, 0, NULL, 0}}; |
| 115 | + /* Parse the command line options */ |
| 116 | + while (1) { |
| 117 | + int cmdOpt; |
| 118 | + int optIdx; |
| 119 | + cmdOpt = getopt_long(argc, argv, "sft:", longOpt, &optIdx); |
| 120 | + if (cmdOpt == -1) break; |
| 121 | + |
| 122 | + /* Invalid option */ |
| 123 | + if (cmdOpt == '?') break; |
| 124 | + |
| 125 | + if (cmdOpt == 's') { |
| 126 | + db_host = optarg; |
| 127 | + } |
| 128 | + if (cmdOpt == 'f') { |
| 129 | + file_path = optarg; |
| 130 | + } |
| 131 | + if (cmdOpt == 't') { |
| 132 | + thread_num = atoi(optarg); |
| 133 | + } |
| 134 | + } |
| 135 | + if (file_path == NULL) { |
| 136 | + ta_log_error("No specified import file list\n"); |
| 137 | + return EXIT_FAILURE; |
| 138 | + } |
| 139 | + if (ta_logger_init() != SC_OK) { |
| 140 | + ta_log_error("Failed to init logger\n"); |
| 141 | + return EXIT_FAILURE; |
| 142 | + } |
| 143 | + scylladb_logger_init(); |
| 144 | + worker_threads = malloc(thread_num * sizeof(pthread_t)); |
| 145 | + worker_data = malloc(thread_num * sizeof(db_worker_thread_t)); |
| 146 | + |
| 147 | + db_permanode_thpool_init(&pool); |
| 148 | + /* create the request-handling threads */ |
| 149 | + for (int i = 0; i < thread_num; i++) { |
| 150 | + db_permanode_thpool_init_worker(worker_data + i, &pool, db_host); |
| 151 | + pthread_create(&worker_threads[i], NULL, db_permanode_worker_handler, (void*)&worker_data[i]); |
| 152 | + } |
| 153 | + init_importer_data(&importer_data, &pool, file_path); |
| 154 | + pthread_create(&importer_thread, NULL, (void*)importer_handler, (void*)&importer_data); |
| 155 | + |
| 156 | + pthread_join(importer_thread, NULL); |
| 157 | + |
| 158 | + db_permanode_tpool_wait(&pool); |
| 159 | + free(worker_data); |
| 160 | + free(worker_threads); |
| 161 | + |
| 162 | + scylladb_logger_release(); |
| 163 | + |
| 164 | + return 0; |
| 165 | +} |
0 commit comments