063c4179b · cryptotaxi247/netdata-1

@@ -95,11 +95,11 @@ struct metadata_cmd {

95 enum metadata_opcode opcode;

96 struct completion *completion;

97 const void *param[MAX_PARAM_LIST];

98 + struct metadata_cmd *prev, *next;

99 };

100

101 struct metadata_database_cmdqueue {

101 - unsigned head, tail;

102 - struct metadata_cmd cmd_array[METADATA_CMD_Q_MAX_SIZE];

102 + struct metadata_cmd *cmd_base;

103 };

104

105 typedef enum {

@@ -120,7 +120,6 @@ struct metadata_wc {

120 struct completion *scan_complete;

121 /* FIFO command queue */

122 uv_mutex_t cmd_mutex;

123 - uv_cond_t cmd_cond;

123 struct metadata_database_cmdqueue cmd_queue;

124 };

125

@@ -1062,103 +1061,57 @@ static void cleanup_health_log(struct metadata_wc *wc)

1061

1062 static void metadata_init_cmd_queue(struct metadata_wc *wc)

1063 {

1065 - wc->cmd_queue.head = wc->cmd_queue.tail = 0;

1066 - wc->queue_size = 0;

1067 - fatal_assert(0 == uv_cond_init(&wc->cmd_cond));

1064 + wc->cmd_queue.cmd_base = NULL;

1065 fatal_assert(0 == uv_mutex_init(&wc->cmd_mutex));

1066 }

1067

1071 -int metadata_enq_cmd_noblock(struct metadata_wc *wc, struct metadata_cmd *cmd)

1068 +static void metadata_free_cmd_queue(struct metadata_wc *wc)

1069 {

1073 - unsigned queue_size;

1074 -

1075 - /* wait for free space in queue */

1070 uv_mutex_lock(&wc->cmd_mutex);

1077 -

1078 - if (cmd->opcode == METADATA_SYNC_SHUTDOWN) {

1079 - metadata_flag_set(wc, METADATA_FLAG_SHUTDOWN);

1080 - uv_mutex_unlock(&wc->cmd_mutex);

1081 - return 0;

1071 + while(wc->cmd_queue.cmd_base) {

1072 + struct metadata_cmd *t = wc->cmd_queue.cmd_base;

1073 + DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(wc->cmd_queue.cmd_base, t, prev, next);

1074 + freez(t);

1075 }

1083 -

1084 - if (unlikely((queue_size = wc->queue_size) == METADATA_CMD_Q_MAX_SIZE ||

1085 - metadata_flag_check(wc, METADATA_FLAG_SHUTDOWN))) {

1086 - uv_mutex_unlock(&wc->cmd_mutex);

1087 - return 1;

1088 - }

1089 -

1090 - fatal_assert(queue_size < METADATA_CMD_Q_MAX_SIZE);

1091 - /* enqueue command */

1092 - wc->cmd_queue.cmd_array[wc->cmd_queue.tail] = *cmd;

1093 - wc->cmd_queue.tail = wc->cmd_queue.tail != METADATA_CMD_Q_MAX_SIZE - 1 ?

1094 - wc->cmd_queue.tail + 1 : 0;

1095 - wc->queue_size = queue_size + 1;

1076 uv_mutex_unlock(&wc->cmd_mutex);

1097 - return 0;

1077 }

1078

1079 static void metadata_enq_cmd(struct metadata_wc *wc, struct metadata_cmd *cmd)

1080 {

1102 - unsigned queue_size;

1103 -

1104 - /* wait for free space in queue */

1105 - uv_mutex_lock(&wc->cmd_mutex);

1106 - if (unlikely(metadata_flag_check(wc, METADATA_FLAG_SHUTDOWN))) {

1107 - uv_mutex_unlock(&wc->cmd_mutex);

1108 - (void) uv_async_send(&wc->async);

1109 - return;

1110 - }

1111 -

1081 if (cmd->opcode == METADATA_SYNC_SHUTDOWN) {

1082 metadata_flag_set(wc, METADATA_FLAG_SHUTDOWN);

1114 - uv_mutex_unlock(&wc->cmd_mutex);

1115 - (void) uv_async_send(&wc->async);

1116 - return;

1083 + goto wakeup_event_loop;

1084 }

1085

1119 - while ((queue_size = wc->queue_size) == METADATA_CMD_Q_MAX_SIZE) {

1120 - if (unlikely(metadata_flag_check(wc, METADATA_FLAG_SHUTDOWN))) {

1121 - uv_mutex_unlock(&wc->cmd_mutex);

1122 - return;

1123 - }

1124 - uv_cond_wait(&wc->cmd_cond, &wc->cmd_mutex);

1125 - }

1126 - fatal_assert(queue_size < METADATA_CMD_Q_MAX_SIZE);

1127 - /* enqueue command */

1128 - wc->cmd_queue.cmd_array[wc->cmd_queue.tail] = *cmd;

1129 - wc->cmd_queue.tail = wc->cmd_queue.tail != METADATA_CMD_Q_MAX_SIZE - 1 ?

1130 - wc->cmd_queue.tail + 1 : 0;

1131 - wc->queue_size = queue_size + 1;

1086 + if (unlikely(metadata_flag_check(wc, METADATA_FLAG_SHUTDOWN)))

1087 + goto wakeup_event_loop;

1088 +

1089 + struct metadata_cmd *t = mallocz(sizeof(*t));

1090 + *t = *cmd;

1091 + t->prev = t->next = NULL;

1092 +

1093 + uv_mutex_lock(&wc->cmd_mutex);

1094 + DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(wc->cmd_queue.cmd_base, t, prev, next);

1095 uv_mutex_unlock(&wc->cmd_mutex);

1096

1134 - /* wake up event loop */

1097 +wakeup_event_loop:

1098 (void) uv_async_send(&wc->async);

1099 }

1100

1101 static struct metadata_cmd metadata_deq_cmd(struct metadata_wc *wc)

1102 {

1103 struct metadata_cmd ret;

1141 - unsigned queue_size;

1104

1105 uv_mutex_lock(&wc->cmd_mutex);

1144 - queue_size = wc->queue_size;

1145 - if (queue_size == 0) {

1146 - memset(&ret, 0, sizeof(ret));

1106 + if(wc->cmd_queue.cmd_base) {

1107 + struct metadata_cmd *t = wc->cmd_queue.cmd_base;

1108 + DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(wc->cmd_queue.cmd_base, t, prev, next);

1109 + ret = *t;

1110 + freez(t);

1111 + }

1112 + else {

1113 ret.opcode = METADATA_DATABASE_NOOP;

1114 ret.completion = NULL;

1149 - } else {

1150 - /* dequeue command */

1151 - ret = wc->cmd_queue.cmd_array[wc->cmd_queue.head];

1152 -

1153 - if (queue_size == 1) {

1154 - wc->cmd_queue.head = wc->cmd_queue.tail = 0;

1155 - } else {

1156 - wc->cmd_queue.head = wc->cmd_queue.head != METADATA_CMD_Q_MAX_SIZE - 1 ?

1157 - wc->cmd_queue.head + 1 : 0;

1158 - }

1159 - wc->queue_size = queue_size - 1;

1160 - /* wake up producers */

1161 - uv_cond_signal(&wc->cmd_cond);

1115 }

1116 uv_mutex_unlock(&wc->cmd_mutex);

1117

@@ -1187,8 +1140,7 @@ static void timer_cb(uv_timer_t* handle)

1140

1141 if (wc->metadata_check_after && wc->metadata_check_after < now) {

1142 cmd.opcode = METADATA_SCAN_HOSTS;

1190 - if (!metadata_enq_cmd_noblock(wc, &cmd))

1191 - wc->metadata_check_after = now + METADATA_HOST_CHECK_INTERVAL;

1143 + metadata_enq_cmd(wc, &cmd);

1144 }

1145 }

1146

@@ -1794,7 +1746,6 @@ static void metadata_event_loop(void *arg)

1746 uv_close((uv_handle_t *)&wc->timer_req, NULL);

1747

1748 uv_close((uv_handle_t *)&wc->async, NULL);

1797 - uv_cond_destroy(&wc->cmd_cond);

1749 int rc;

1750 do {

1751 rc = uv_loop_close(loop);

@@ -1808,6 +1759,7 @@ static void metadata_event_loop(void *arg)

1759 completion_mark_complete(&wc->init_complete);

1760 completion_destroy(wc->scan_complete);

1761 freez(wc->scan_complete);

1762 + metadata_free_cmd_queue(wc);

1763 return;

1764

1765 error_after_timer_init:

@@ -2024,7 +1976,6 @@ int metadata_unittest(void)

1976 // Queue items for a specific period of time

1977 metadata_unittest_threads();

1978

2027 - fprintf(stderr, "Items still in queue %u\n", metasync_worker.queue_size);

1979 metadata_sync_shutdown();

1980

1981 return 0;

dynamic meta queue size (#16218)