1/*
2 * storage.c
3 *
4 * Copyright (C) 2009-2016 Aerospike, Inc.
5 *
6 * Portions may be licensed to Aerospike, Inc. under one or more contributor
7 * license agreements.
8 *
9 * This program is free software: you can redistribute it and/or modify it under
10 * the terms of the GNU Affero General Public License as published by the Free
11 * Software Foundation, either version 3 of the License, or (at your option) any
12 * later version.
13 *
14 * This program is distributed in the hope that it will be useful, but WITHOUT
15 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
16 * FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
17 * details.
18 *
19 * You should have received a copy of the GNU Affero General Public License
20 * along with this program. If not, see http://www.gnu.org/licenses/
21 */
22
23//==========================================================
24// Includes.
25//
26
27#include "storage/storage.h"
28
29#include <stdbool.h>
30#include <stddef.h>
31#include <stdint.h>
32#include <string.h>
33
34#include "citrusleaf/cf_digest.h"
35#include "citrusleaf/cf_queue.h"
36
37#include "cf_mutex.h"
38#include "fault.h"
39
40#include "base/cfg.h"
41#include "base/datamodel.h"
42#include "base/index.h"
43#include "base/thr_info.h"
44#include "fabric/partition.h"
45
46
47//==========================================================
48// Generic "base class" functions that call through
49// storage-engine "v-tables".
50//
51
52//--------------------------------------
53// as_storage_cfg_init
54//
55
56typedef void (*as_storage_cfg_init_fn)(as_namespace *ns);
57static const as_storage_cfg_init_fn as_storage_cfg_init_table[AS_NUM_STORAGE_ENGINES] = {
58 NULL, // memory doesn't need this
59 as_storage_cfg_init_ssd
60};
61
62void
63as_storage_cfg_init(as_namespace *ns)
64{
65 if (as_storage_cfg_init_table[ns->storage_type]) {
66 as_storage_cfg_init_table[ns->storage_type](ns);
67 }
68}
69
70//--------------------------------------
71// as_storage_init
72//
73
74typedef void (*as_storage_namespace_init_fn)(as_namespace *ns);
75static const as_storage_namespace_init_fn as_storage_namespace_init_table[AS_NUM_STORAGE_ENGINES] = {
76 as_storage_namespace_init_memory,
77 as_storage_namespace_init_ssd
78};
79
80void
81as_storage_init()
82{
83 // Includes resuming indexes for warm and cool restarts.
84
85 for (uint32_t i = 0; i < g_config.n_namespaces; i++) {
86 as_namespace *ns = g_config.namespaces[i];
87
88 if (as_storage_namespace_init_table[ns->storage_type]) {
89 as_storage_namespace_init_table[ns->storage_type](ns);
90 }
91 }
92}
93
94//--------------------------------------
95// as_storage_load
96//
97
98typedef void (*as_storage_namespace_load_fn)(as_namespace *ns, cf_queue *complete_q);
99static const as_storage_namespace_load_fn as_storage_namespace_load_table[AS_NUM_STORAGE_ENGINES] = {
100 NULL, // memory has no load phase
101 as_storage_namespace_load_ssd
102};
103
104#define TICKER_INTERVAL (5 * 1000) // 5 seconds
105
106void
107as_storage_load()
108{
109 // Includes device scans for cold starts and cool restarts.
110
111 cf_queue complete_q;
112
113 cf_queue_init(&complete_q, sizeof(void*), g_config.n_namespaces, true);
114
115 for (uint32_t i = 0; i < g_config.n_namespaces; i++) {
116 as_namespace *ns = g_config.namespaces[i];
117
118 if (as_storage_namespace_load_table[ns->storage_type]) {
119 as_storage_namespace_load_table[ns->storage_type](ns, &complete_q);
120 }
121 else {
122 void *_t = NULL;
123
124 cf_queue_push(&complete_q, &_t);
125 }
126 }
127
128 // Wait for completion - cold starts or cool restarts may take a while.
129
130 for (uint32_t i = 0; i < g_config.n_namespaces; i++) {
131 void *_t;
132
133 while (cf_queue_pop(&complete_q, &_t, TICKER_INTERVAL) != CF_QUEUE_OK) {
134 as_storage_loading_records_ticker_ssd();
135 }
136 }
137
138 cf_queue_destroy(&complete_q);
139}
140
141//--------------------------------------
142// as_storage_start_tomb_raider
143//
144
145typedef void (*as_storage_start_tomb_raider_fn)(as_namespace *ns);
146static const as_storage_start_tomb_raider_fn as_storage_start_tomb_raider_table[AS_NUM_STORAGE_ENGINES] = {
147 as_storage_start_tomb_raider_memory,
148 as_storage_start_tomb_raider_ssd
149};
150
151void
152as_storage_start_tomb_raider()
153{
154 for (uint32_t i = 0; i < g_config.n_namespaces; i++) {
155 as_namespace *ns = g_config.namespaces[i];
156
157 if (as_storage_start_tomb_raider_table[ns->storage_type]) {
158 as_storage_start_tomb_raider_table[ns->storage_type](ns);
159 }
160 }
161}
162
163//--------------------------------------
164// as_storage_namespace_destroy
165//
166
167typedef int (*as_storage_namespace_destroy_fn)(as_namespace *ns);
168static const as_storage_namespace_destroy_fn as_storage_namespace_destroy_table[AS_NUM_STORAGE_ENGINES] = {
169 NULL, // memory has no destroy
170 as_storage_namespace_destroy_ssd
171};
172
173int
174as_storage_namespace_destroy(as_namespace *ns)
175{
176 if (as_storage_namespace_destroy_table[ns->storage_type]) {
177 return as_storage_namespace_destroy_table[ns->storage_type](ns);
178 }
179
180 return 0;
181}
182
183//--------------------------------------
184// as_storage_record_destroy
185//
186
187typedef int (*as_storage_record_destroy_fn)(as_namespace *ns, as_record *r);
188static const as_storage_record_destroy_fn as_storage_record_destroy_table[AS_NUM_STORAGE_ENGINES] = {
189 NULL, // memory has no record destroy
190 as_storage_record_destroy_ssd
191};
192
193int
194as_storage_record_destroy(as_namespace *ns, as_record *r)
195{
196 if (as_storage_record_destroy_table[ns->storage_type]) {
197 return as_storage_record_destroy_table[ns->storage_type](ns, r);
198 }
199
200 return 0;
201}
202
203//--------------------------------------
204// as_storage_record_create
205//
206
207typedef int (*as_storage_record_create_fn)(as_storage_rd *rd);
208static const as_storage_record_create_fn as_storage_record_create_table[AS_NUM_STORAGE_ENGINES] = {
209 NULL, // memory has no record create
210 as_storage_record_create_ssd
211};
212
213int
214as_storage_record_create(as_namespace *ns, as_record *r, as_storage_rd *rd)
215{
216 rd->r = r;
217 rd->ns = ns;
218 rd->bins = 0;
219 rd->n_bins = 0;
220 rd->record_on_device = false;
221 rd->ignore_record_on_device = false;
222 rd->set_name_len = 0;
223 rd->set_name = NULL;
224 rd->key_size = 0;
225 rd->key = NULL;
226 rd->read_page_cache = false;
227 rd->is_durable_delete = false;
228 rd->keep_pickle = false;
229 rd->pickle_sz = 0;
230 rd->orig_pickle_sz = 0;
231 rd->pickle = NULL;
232
233 if (as_storage_record_create_table[ns->storage_type]) {
234 return as_storage_record_create_table[ns->storage_type](rd);
235 }
236
237 return 0;
238}
239
240//--------------------------------------
241// as_storage_record_open
242//
243
244typedef int (*as_storage_record_open_fn)(as_storage_rd *rd);
245static const as_storage_record_open_fn as_storage_record_open_table[AS_NUM_STORAGE_ENGINES] = {
246 NULL, // memory has no record open
247 as_storage_record_open_ssd
248};
249
250int
251as_storage_record_open(as_namespace *ns, as_record *r, as_storage_rd *rd)
252{
253 rd->r = r;
254 rd->ns = ns;
255 rd->bins = 0;
256 rd->n_bins = 0;
257 rd->record_on_device = true;
258 rd->ignore_record_on_device = false;
259 rd->set_name_len = 0;
260 rd->set_name = NULL;
261 rd->key_size = 0;
262 rd->key = NULL;
263 rd->read_page_cache = false;
264 rd->is_durable_delete = false;
265 rd->keep_pickle = false;
266 rd->pickle_sz = 0;
267 rd->orig_pickle_sz = 0;
268 rd->pickle = NULL;
269
270 if (as_storage_record_open_table[ns->storage_type]) {
271 return as_storage_record_open_table[ns->storage_type](rd);
272 }
273
274 return 0;
275}
276
277//--------------------------------------
278// as_storage_record_close
279//
280
281typedef int (*as_storage_record_close_fn)(as_storage_rd *rd);
282static const as_storage_record_close_fn as_storage_record_close_table[AS_NUM_STORAGE_ENGINES] = {
283 NULL, // memory has no record close
284 as_storage_record_close_ssd
285};
286
287int
288as_storage_record_close(as_storage_rd *rd)
289{
290 if (as_storage_record_close_table[rd->ns->storage_type]) {
291 return as_storage_record_close_table[rd->ns->storage_type](rd);
292 }
293
294 return 0;
295}
296
297//--------------------------------------
298// as_storage_record_load_n_bins
299//
300
301typedef int (*as_storage_record_load_n_bins_fn)(as_storage_rd *rd);
302static const as_storage_record_load_n_bins_fn as_storage_record_load_n_bins_table[AS_NUM_STORAGE_ENGINES] = {
303 NULL, // memory has no record load n bins
304 as_storage_record_load_n_bins_ssd
305};
306
307int
308as_storage_record_load_n_bins(as_storage_rd *rd)
309{
310 if (as_storage_record_load_n_bins_table[rd->ns->storage_type]) {
311 return as_storage_record_load_n_bins_table[rd->ns->storage_type](rd);
312 }
313
314 return 0;
315}
316
317//--------------------------------------
318// as_storage_record_load_bins
319//
320
321typedef int (*as_storage_record_load_bins_fn)(as_storage_rd *rd);
322static const as_storage_record_load_bins_fn as_storage_record_load_bins_table[AS_NUM_STORAGE_ENGINES] = {
323 NULL, // memory has no record load bins
324 as_storage_record_load_bins_ssd
325};
326
327int
328as_storage_record_load_bins(as_storage_rd *rd)
329{
330 if (as_storage_record_load_bins_table[rd->ns->storage_type]) {
331 return as_storage_record_load_bins_table[rd->ns->storage_type](rd);
332 }
333
334 return 0;
335}
336
337//--------------------------------------
338// as_storage_record_size_and_check
339//
340
341typedef bool (*as_storage_record_size_and_check_fn)(as_storage_rd *rd);
342static const as_storage_record_size_and_check_fn as_storage_record_size_and_check_table[AS_NUM_STORAGE_ENGINES] = {
343 NULL, // no limit if no persistent storage - flat size is irrelevant
344 as_storage_record_size_and_check_ssd
345};
346
347bool
348as_storage_record_size_and_check(as_storage_rd *rd)
349{
350 if (as_storage_record_size_and_check_table[rd->ns->storage_type]) {
351 return as_storage_record_size_and_check_table[rd->ns->storage_type](rd);
352 }
353
354 return true;
355}
356
357//--------------------------------------
358// as_storage_record_write
359//
360
361typedef int (*as_storage_record_write_fn)(as_storage_rd *rd);
362static const as_storage_record_write_fn as_storage_record_write_table[AS_NUM_STORAGE_ENGINES] = {
363 as_storage_record_write_memory,
364 as_storage_record_write_ssd
365};
366
367int
368as_storage_record_write(as_storage_rd *rd)
369{
370 if (as_storage_record_write_table[rd->ns->storage_type]) {
371 return as_storage_record_write_table[rd->ns->storage_type](rd);
372 }
373
374 return 0;
375}
376
377//--------------------------------------
378// as_storage_wait_for_defrag
379//
380
381typedef void (*as_storage_wait_for_defrag_fn)(as_namespace *ns);
382static const as_storage_wait_for_defrag_fn as_storage_wait_for_defrag_table[AS_NUM_STORAGE_ENGINES] = {
383 NULL, // memory doesn't do defrag
384 as_storage_wait_for_defrag_ssd
385};
386
387void
388as_storage_wait_for_defrag()
389{
390 for (uint32_t i = 0; i < g_config.n_namespaces; i++) {
391 as_namespace *ns = g_config.namespaces[i];
392
393 if (as_storage_wait_for_defrag_table[ns->storage_type]) {
394 as_storage_wait_for_defrag_table[ns->storage_type](ns);
395 }
396 }
397}
398
399//--------------------------------------
400// as_storage_overloaded
401//
402
403typedef bool (*as_storage_overloaded_fn)(as_namespace *ns);
404static const as_storage_overloaded_fn as_storage_overloaded_table[AS_NUM_STORAGE_ENGINES] = {
405 NULL, // memory has no overload check
406 as_storage_overloaded_ssd
407};
408
409bool
410as_storage_overloaded(as_namespace *ns)
411{
412 if (as_storage_overloaded_table[ns->storage_type]) {
413 return as_storage_overloaded_table[ns->storage_type](ns);
414 }
415
416 return false;
417}
418
419//--------------------------------------
420// as_storage_has_space
421//
422
423typedef bool (*as_storage_has_space_fn)(as_namespace *ns);
424static const as_storage_has_space_fn as_storage_has_space_table[AS_NUM_STORAGE_ENGINES] = {
425 NULL, // memory has no space check
426 as_storage_has_space_ssd
427};
428
429bool
430as_storage_has_space(as_namespace *ns)
431{
432 if (as_storage_has_space_table[ns->storage_type]) {
433 return as_storage_has_space_table[ns->storage_type](ns);
434 }
435
436 return true;
437}
438
439//--------------------------------------
440// as_storage_defrag_sweep
441//
442
443typedef void (*as_storage_defrag_sweep_fn)(as_namespace *ns);
444static const as_storage_defrag_sweep_fn as_storage_defrag_sweep_table[AS_NUM_STORAGE_ENGINES] = {
445 NULL, // memory doesn't do defrag
446 as_storage_defrag_sweep_ssd
447};
448
449void
450as_storage_defrag_sweep(as_namespace *ns)
451{
452 if (as_storage_defrag_sweep_table[ns->storage_type]) {
453 as_storage_defrag_sweep_table[ns->storage_type](ns);
454 }
455}
456
457//--------------------------------------
458// as_storage_load_regime
459//
460
461typedef void (*as_storage_load_regime_fn)(as_namespace *ns);
462static const as_storage_load_regime_fn as_storage_load_regime_table[AS_NUM_STORAGE_ENGINES] = {
463 NULL, // memory doesn't store info
464 as_storage_load_regime_ssd
465};
466
467void
468as_storage_load_regime(as_namespace *ns)
469{
470 if (as_storage_load_regime_table[ns->storage_type]) {
471 as_storage_load_regime_table[ns->storage_type](ns);
472 }
473}
474
475//--------------------------------------
476// as_storage_save_regime
477//
478
479typedef void (*as_storage_save_regime_fn)(as_namespace *ns);
480static const as_storage_save_regime_fn as_storage_save_regime_table[AS_NUM_STORAGE_ENGINES] = {
481 NULL, // memory doesn't store info
482 as_storage_save_regime_ssd
483};
484
485void
486as_storage_save_regime(as_namespace *ns)
487{
488 if (as_storage_save_regime_table[ns->storage_type]) {
489 as_storage_save_regime_table[ns->storage_type](ns);
490 }
491}
492
493//--------------------------------------
494// as_storage_load_roster_generation
495//
496
497typedef void (*as_storage_load_roster_generation_fn)(as_namespace *ns);
498static const as_storage_load_roster_generation_fn as_storage_load_roster_generation_table[AS_NUM_STORAGE_ENGINES] = {
499 NULL, // memory doesn't store info
500 as_storage_load_roster_generation_ssd
501};
502
503void
504as_storage_load_roster_generation(as_namespace *ns)
505{
506 if (as_storage_load_roster_generation_table[ns->storage_type]) {
507 as_storage_load_roster_generation_table[ns->storage_type](ns);
508 }
509}
510
511//--------------------------------------
512// as_storage_save_roster_generation
513//
514
515typedef void (*as_storage_save_roster_generation_fn)(as_namespace *ns);
516static const as_storage_save_roster_generation_fn as_storage_save_roster_generation_table[AS_NUM_STORAGE_ENGINES] = {
517 NULL, // memory doesn't store info
518 as_storage_save_roster_generation_ssd
519};
520
521void
522as_storage_save_roster_generation(as_namespace *ns)
523{
524 if (as_storage_save_roster_generation_table[ns->storage_type]) {
525 as_storage_save_roster_generation_table[ns->storage_type](ns);
526 }
527}
528
529//--------------------------------------
530// as_storage_load_pmeta
531//
532
533typedef void (*as_storage_load_pmeta_fn)(as_namespace *ns, as_partition *p);
534static const as_storage_load_pmeta_fn as_storage_load_pmeta_table[AS_NUM_STORAGE_ENGINES] = {
535 as_storage_load_pmeta_memory,
536 as_storage_load_pmeta_ssd
537};
538
539void
540as_storage_load_pmeta(as_namespace *ns, as_partition *p)
541{
542 if (as_storage_load_pmeta_table[ns->storage_type]) {
543 as_storage_load_pmeta_table[ns->storage_type](ns, p);
544 }
545}
546
547//--------------------------------------
548// as_storage_save_pmeta
549//
550
551typedef void (*as_storage_save_pmeta_fn)(as_namespace *ns, const as_partition *p);
552static const as_storage_save_pmeta_fn as_storage_save_pmeta_table[AS_NUM_STORAGE_ENGINES] = {
553 NULL, // memory doesn't support info
554 as_storage_save_pmeta_ssd
555};
556
557void
558as_storage_save_pmeta(as_namespace *ns, const as_partition *p)
559{
560 if (as_storage_save_pmeta_table[ns->storage_type]) {
561 as_storage_save_pmeta_table[ns->storage_type](ns, p);
562 }
563}
564
565//--------------------------------------
566// as_storage_cache_pmeta
567//
568
569typedef void (*as_storage_cache_pmeta_fn)(as_namespace *ns, const as_partition *p);
570static const as_storage_cache_pmeta_fn as_storage_cache_pmeta_table[AS_NUM_STORAGE_ENGINES] = {
571 NULL, // memory doesn't support info
572 as_storage_cache_pmeta_ssd
573};
574
575void
576as_storage_cache_pmeta(as_namespace *ns, const as_partition *p)
577{
578 if (as_storage_cache_pmeta_table[ns->storage_type]) {
579 as_storage_cache_pmeta_table[ns->storage_type](ns, p);
580 }
581}
582
583//--------------------------------------
584// as_storage_flush_pmeta
585//
586
587typedef void (*as_storage_flush_pmeta_fn)(as_namespace *ns, uint32_t start_pid, uint32_t n_partitions);
588static const as_storage_flush_pmeta_fn as_storage_flush_pmeta_table[AS_NUM_STORAGE_ENGINES] = {
589 NULL, // memory doesn't support info
590 as_storage_flush_pmeta_ssd
591};
592
593void
594as_storage_flush_pmeta(as_namespace *ns, uint32_t start_pid, uint32_t n_partitions)
595{
596 if (as_storage_flush_pmeta_table[ns->storage_type]) {
597 as_storage_flush_pmeta_table[ns->storage_type](ns, start_pid, n_partitions);
598 }
599}
600
601//--------------------------------------
602// as_storage_stats
603//
604
605typedef int (*as_storage_stats_fn)(as_namespace *ns, int *available_pct, uint64_t *used_disk_bytes);
606static const as_storage_stats_fn as_storage_stats_table[AS_NUM_STORAGE_ENGINES] = {
607 as_storage_stats_memory,
608 as_storage_stats_ssd
609};
610
611int
612as_storage_stats(as_namespace *ns, int *available_pct, uint64_t *used_disk_bytes)
613{
614 if (as_storage_stats_table[ns->storage_type]) {
615 return as_storage_stats_table[ns->storage_type](ns, available_pct, used_disk_bytes);
616 }
617
618 return 0;
619}
620
621//--------------------------------------
622// as_storage_device_stats
623//
624
625typedef void (*as_storage_device_stats_fn)(as_namespace *ns, uint32_t device_ix, storage_device_stats *stats);
626static const as_storage_device_stats_fn as_storage_device_stats_table[AS_NUM_STORAGE_ENGINES] = {
627 NULL,
628 as_storage_device_stats_ssd
629};
630
631void
632as_storage_device_stats(as_namespace *ns, uint32_t device_ix, storage_device_stats *stats)
633{
634 if (as_storage_device_stats_table[ns->storage_type]) {
635 as_storage_device_stats_table[ns->storage_type](ns, device_ix, stats);
636 return;
637 }
638
639 memset(stats, 0, sizeof(storage_device_stats));
640}
641
642//--------------------------------------
643// as_storage_ticker_stats
644//
645
646typedef int (*as_storage_ticker_stats_fn)(as_namespace *ns);
647static const as_storage_ticker_stats_fn as_storage_ticker_stats_table[AS_NUM_STORAGE_ENGINES] = {
648 NULL, // memory doesn't support per-disk histograms... for now.
649 as_storage_ticker_stats_ssd
650};
651
652int
653as_storage_ticker_stats(as_namespace *ns)
654{
655 if (as_storage_ticker_stats_table[ns->storage_type]) {
656 return as_storage_ticker_stats_table[ns->storage_type](ns);
657 }
658
659 return 0;
660}
661
662//--------------------------------------
663// as_storage_histogram_clear_all
664//
665
666typedef int (*as_storage_histogram_clear_fn)(as_namespace *ns);
667static const as_storage_histogram_clear_fn as_storage_histogram_clear_table[AS_NUM_STORAGE_ENGINES] = {
668 NULL, // memory doesn't support per-disk histograms... for now.
669 as_storage_histogram_clear_ssd
670};
671
672int
673as_storage_histogram_clear_all(as_namespace *ns)
674{
675 if (as_storage_histogram_clear_table[ns->storage_type]) {
676 return as_storage_histogram_clear_table[ns->storage_type](ns);
677 }
678
679 return 0;
680}
681
682//--------------------------------------
683// as_storage_record_size
684//
685
686typedef uint32_t (*as_storage_record_size_fn)(const as_record *r);
687static const as_storage_record_size_fn as_storage_record_size_table[AS_NUM_STORAGE_ENGINES] = {
688 NULL, // memory doesn't support record stored size.
689 as_storage_record_size_ssd
690};
691
692uint32_t
693as_storage_record_size(const as_namespace *ns, const as_record *r)
694{
695 if (as_storage_record_size_table[ns->storage_type]) {
696 return as_storage_record_size_table[ns->storage_type](r);
697 }
698
699 return 0;
700}
701
702
703//==========================================================
704// Generic functions that don't use "v-tables".
705//
706
707// Get size of record's in-memory data - everything except index bytes.
708uint64_t
709as_storage_record_get_n_bytes_memory(as_storage_rd *rd)
710{
711 if (! rd->ns->storage_data_in_memory) {
712 return 0;
713 }
714
715 uint64_t n_bytes_memory = 0;
716
717 for (uint16_t i = 0; i < rd->n_bins; i++) {
718 n_bytes_memory += as_bin_particle_size(&rd->bins[i]);
719 }
720
721 if (! rd->ns->single_bin) {
722 if (rd->r->key_stored == 1) {
723 n_bytes_memory += sizeof(as_rec_space) +
724 ((as_rec_space*)rd->r->dim)->key_size;
725 }
726
727 if (as_index_get_bin_space(rd->r)) {
728 n_bytes_memory += sizeof(as_bin_space) +
729 (sizeof(as_bin) * rd->n_bins);
730 }
731 }
732
733 return n_bytes_memory;
734}
735
736void
737as_storage_record_adjust_mem_stats(as_storage_rd *rd, uint64_t start_bytes)
738{
739 if (! rd->ns->storage_data_in_memory) {
740 return;
741 }
742
743 uint64_t end_bytes = as_storage_record_get_n_bytes_memory(rd);
744 int64_t delta_bytes = (int64_t)end_bytes - (int64_t)start_bytes;
745
746 if (delta_bytes != 0) {
747 cf_atomic_int_add(&rd->ns->n_bytes_memory, delta_bytes);
748 as_namespace_adjust_set_memory(rd->ns, as_index_get_set_id(rd->r),
749 delta_bytes);
750 }
751}
752
753void
754as_storage_record_drop_from_mem_stats(as_storage_rd *rd)
755{
756 if (! rd->ns->storage_data_in_memory) {
757 return;
758 }
759
760 uint64_t drop_bytes = as_storage_record_get_n_bytes_memory(rd);
761
762 cf_atomic_int_sub(&rd->ns->n_bytes_memory, drop_bytes);
763 as_namespace_adjust_set_memory(rd->ns, as_index_get_set_id(rd->r),
764 -(int64_t)drop_bytes);
765}
766
767void
768as_storage_record_get_set_name(as_storage_rd *rd)
769{
770 rd->set_name = as_index_get_set_name(rd->r, rd->ns);
771
772 if (rd->set_name) {
773 rd->set_name_len = strlen(rd->set_name);
774 }
775}
776
777bool
778as_storage_record_get_key(as_storage_rd *rd)
779{
780 if (rd->r->key_stored == 0) {
781 return false;
782 }
783
784 if (rd->ns->storage_data_in_memory) {
785 rd->key_size = ((as_rec_space*)rd->r->dim)->key_size;
786 rd->key = ((as_rec_space*)rd->r->dim)->key;
787 return true;
788 }
789
790 if (rd->record_on_device && ! rd->ignore_record_on_device) {
791 return as_storage_record_get_key_ssd(rd);
792 }
793
794 return false;
795}
796
797bool
798as_storage_record_get_pickle(as_storage_rd *rd)
799{
800 if (rd->ns->storage_data_in_memory) {
801 as_storage_record_get_set_name(rd);
802 as_storage_record_get_key(rd);
803 as_storage_rd_load_n_bins(rd);
804 as_storage_rd_load_bins(rd, NULL);
805 as_flat_pickle_record(rd);
806 return true;
807 }
808
809 return as_storage_record_get_pickle_ssd(rd);
810}
811
812void
813as_storage_shutdown(uint32_t instance)
814{
815 for (uint32_t i = 0; i < g_config.n_namespaces; i++) {
816 as_namespace *ns = g_config.namespaces[i];
817
818 if (ns->storage_type == AS_STORAGE_ENGINE_SSD) {
819 // Lock all record locks - stops everything writing to current swbs
820 // such that each write's record lock scope is either completed or
821 // never entered.
822 for (uint32_t pid = 0; pid < AS_PARTITIONS; pid++) {
823 as_partition_shutdown(ns, pid);
824 }
825
826 cf_info(AS_STORAGE, "{%s} partitions shut down", ns->name);
827
828 // Now flush everything outstanding to storage devices.
829 as_storage_shutdown_ssd(ns);
830
831 cf_info(AS_STORAGE, "{%s} storage devices flushed", ns->name);
832
833 as_namespace_xmem_shutdown(ns, instance);
834 }
835 else {
836 cf_info(AS_STORAGE, "{%s} storage-engine memory - nothing to do",
837 ns->name);
838 }
839 }
840}
841