appendb/disk_read_ops.cpp at master · lockerdome/appendb · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
#include "drive.hpp"
#include "block_index.hpp"
#include "read_ops.hpp"
#include <map>
#include <vector>

namespace AppenDB {
  // TODO: avoid blocking the main thread on a page fault
  // TODO: use a better datastructure

  struct ResultSegment {
    uint64_t begin;
    uint32_t length;
    bool is_free;
    ResultSegment(uint64_t b, uint32_t l) : begin(b), length(l), is_free(false) {}
    ~ResultSegment() {}
  };

  struct Result {
    ::std::vector<ResultSegment> segments;
  };

  static ::std::map<Drive::UserData, Result> results;

  Drive::ResultSegmentCount Drive::result_num_segments(Drive::UserData result_id) const {
    ::std::map<Drive::UserData, Result>::const_iterator it = results.find(result_id);
    assert(it != results.end());
    return it->second.segments.size();
  }

  void Drive::result_read_segment(Drive::UserData result_id, Drive::ResultSegmentCount segment_id, const char *& segment_begin, size_t &segment_length) const {
    ::std::map<Drive::UserData, Result>::const_iterator it = results.find(result_id);
    assert(it != results.end());
    const Result& result = it->second;
    assert(result.segments.size() > segment_id);
    const ResultSegment &segment = result.segments[segment_id];
    assert(!segment.is_free);
    segment_begin = disk() + segment.begin;
    segment_length = segment.length;
  }

  void Drive::result_free(Drive::UserData result_id) {
    ::std::map<Drive::UserData, Result>::iterator it = results.find(result_id);
    assert(it != results.end());
    Result& result = it->second;
    ::std::vector<ResultSegment>& segments = result.segments;
    for(::std::vector<ResultSegment>::iterator i = segments.begin(); i != segments.end(); ++i){
      if(!(i->is_free)){
        // free segment
        data_range_remove_lock(i->begin, i->length);
      }
    }

    // free result
    results.erase(it);
  }

  void Drive::result_free_segment(Drive::UserData result_id, Drive::ResultSegmentCount segment_id) {
    ::std::map<Drive::UserData, Result>::iterator it = results.find(result_id);
    assert(it != results.end());
    Result& result = it->second;
    assert(result.segments.size() > segment_id);
    ResultSegment &segment = result.segments[segment_id];
    assert(!segment.is_free);
    data_range_remove_lock(segment.begin, segment.length);
    segment.is_free = true;
  }

  void Drive::read_raw_data(UserData result_id, TableIndex type, uint16_t shard, uint32_t time_begin, uint32_t time_end) {
    assert(results.find(result_id) == results.end());
    Result& result = results[result_id];

    block_index(0).for_each_match_range(type, shard, time_begin, time_end, [this, &result](uint32_t block_index){
      size_t const block_start = this->block_data_start(block_index) - this->disk();
      size_t const block_size = this->block_data_size(block_index);

      // TODO: we need to lock this potentially
      // data_range_add_lock(block_start, block_size);

      result.segments.emplace_back(block_start, block_size);
    });

    // TODO: don't do this synchronously
    on_result_ready(result_id);
  }

  void Drive::read_raw_sampled_data(UserData result_id, TableIndex type, uint16_t shard, uint32_t time_begin, uint32_t time_end) {
    // TODO: actually handle sampling data
    read_raw_data(result_id, type, type, time_begin, time_end);
  }

  void Drive::on_result_ready(Drive::ResultID result_id){
    ReadOperation<Drive>::on_result_ready(*this, result_id);
  }

  void Drive::on_result_failure(Drive::ResultID result_id, uint16_t error_code, String error_message){
    ReadOperation<Drive>::on_result_failure(*this, result_id, error_code, error_message);
  }
}