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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
|
/*
* include/framework/scheduling/Epoch.h
*
* Copyright (C) 2023 Douglas B. Rumbaugh <drumbaugh@psu.edu>
* Dong Xie <dongx@psu.edu>
*
* All rights reserved. Published under the Modified BSD License.
*
*/
#pragma once
#include "framework/structure/MutableBuffer.h"
#include "framework/structure/ExtensionStructure.h"
#include "framework/structure/BufferView.h"
namespace de {
template <RecordInterface R, ShardInterface S, QueryInterface Q, LayoutPolicy L>
class Epoch {
private:
typedef MutableBuffer<R> Buffer;
typedef ExtensionStructure<R, S, Q, L> Structure;
typedef BufferView<R, Q> BufView;
public:
Epoch()
: m_buffers()
, m_structure(nullptr)
, m_active_jobs(0)
{}
Epoch(Structure *structure, Buffer *buff)
: m_buffers()
, m_structure(structure)
, m_active_jobs(0)
{
m_buffers.push_back(buff);
}
~Epoch() {
assert(m_active_jobs.load() == 0);
for (auto buf : m_buffers) {
buf.release_reference();
}
if (m_structure) {
m_structure->release_reference();
}
}
void add_buffer(Buffer *buf) {
assert(buf);
buf->take_reference();
m_buffers.push_back(buf);
}
void start_job() {
m_active_jobs.fetch_add(1);
}
void end_job() {
m_active_jobs.fetch_add(-1);
}
size_t get_active_job_num() {
return m_active_jobs.load();
}
Structure *get_structure() {
return m_structure;
}
std::vector<Buffer *> &get_buffers() {
return m_buffers;
}
BufView get_buffer_view() {
return BufView(m_buffers);
}
Buffer *get_active_buffer() {
if (m_buffers.size() == 0) return nullptr;
return m_buffers[m_buffers.size() - 1];
}
/*
* Return the number of buffers in this epoch at
* time of call, and then clear the buffer vector,
* releasing all references in the process.
*/
size_t clear_buffers() {
size_t buf_cnt = m_buffers.size();
for (auto buf : m_buffers) {
if (buf) buf->release_reference();
}
m_buffers.clear();
return buf_cnt;
}
/*
* Returns a new Epoch object that is a copy of this one. The new object will also contain
* a copy of the m_structure, rather than a reference to the same one.
*/
Epoch *clone() {
auto epoch = new Epoch();
epoch->m_buffers = m_buffers;
if (m_structure) {
epoch->m_structure = m_structure->copy();
}
}
private:
Structure *m_structure;
std::vector<Buffer *> m_buffers;
/*
* The number of currently active jobs
* (queries/merges) operating on this
* epoch. An epoch can only be retired
* when this number is 0.
*/
std::atomic<size_t> m_active_jobs;
};
}
|